7 - Communicating with Remote LRCs and the rtiexec

Table of Contents

• 7.1 Time Factory Wrapper
• 7.2 The Connection Manager
• 7.3 The Asynchronous I/O Connection Manager
• 7.4 Changing the RTI's Communications Infrastructure
  • 7.4.1 Subclassing DtSocket
  • 7.4.2 The Role of DtRtiConnection
  • 7.4.3 Overriding Connection Manager Functions
    • Overriding DtConnectionMgr::tick()
  • 7.4.4 Example
• 7.5 Messages

To understand how communication works in the MAK RTI, you need to understand:

• Connections. A connection, represented by the DtRtiConnection class, is a logical connection between the local LRC and other federates' LRCs and the rtiexec (if you are running it). DtRtiConnections are used to send and receive RTI messages.
• Messages. Messages are represented by instances of subclasses of DtRtiMsg (rtiMsg.h).
• The Connection Manager. The DtConnectionMgr is responsible for creating and managing the LRC's connections.

Note

If you are familiar with the VR-Link API, you will see that a DtRtiConnection is analogous to the DtExerciseConn in the DIS version of VR-Link. Similarly, DtRtiMsg and its subclasses closely resemble VR-Link's DtPdu and its subclasses.

7.1 Time Factory Wrapper

The definition of time is different in the 1.3, 1516, and HLA Evolved APIs. Internally, the RTI must use the HLA API for time to handle time management and to process other services that deliver time parameters. The MAK RTI creates a wrapper, DtLogicalTimeFactory, around the different time factories to provide a neutral interface to either implementation.

7.2 The Connection Manager

When a piece of code within the RTI needs to send a message to other federates or the rtiexec, it creates an instance of the appropriate kind of DtRtiMsg, and asks the DtConnectionMgr to send it by calling its sendStamped() or sendStampedAndDelete() function. Based on the value of the transportType argument to these functions (either DtBest_Effort or DtReliable), the DtConnectionMgr sends the message through an appropriate connection. For internal RTI bookkeeping messages (for example, publish and subscribe), the transportType passed to the send functions should be the value returned by DtConnectionMgr::internalTransportType(), which is dictated by your RID settings.

The DtConnectionMgr creates and manages up to two logical connections - a best effort connection and a reliable connection. (If your RID settings indicate that all data should be sent best effort, the reliable connection is not created.) Messages are sent through these connections without regard to how the connections are actually implemented. In other words, just because data is sent through the reliable logical connection, it is not guaranteed that it goes out over a reliable physical connection. Connections are created in the virtual method DtConnectionMgr::initConnections(). Prior to making that call, the Connection Manager invokes the virtual method DtConnectionMgr::initParams() to allow for a separate parameter setup phase of the connection initialization. Both of these calls are made in DtConnectionMgr::init(), which is also virtual.

The DtConnectionMgr functions reliableConn() and udpConn() return pointers to the two connections (reliableConn() could return NULL). The internalConn() function returns a pointer to whichever of these connections is currently being used to send RTI internal bookkeeping data. Although pointers to individual connections are available through these functions, RTI code should always send messages by calling the DtConnectionMgr sending functions rather than by directly using the individual connections. This gives DtConnectionMgr the opportunity to do queueing and buffering if asynchronous I/O is enabled.

There is no separate logical connection to the rtiexec. The rtiexec listens to the same data streams that remote LRCs do, and picks out the messages that it needs to process.

The figure shows how the physical connections used by messages sent through the Connection Manager's logical connections vary depending on the setting for the RTI_fomDataTransportTypeControl and RTI_internalMsgReliableWhenUsingRtiexec parameters in the rid.mtl file.

Configuring the Transport Type

7.3 The Asynchronous I/O Connection Manager

DtAsyncConnectionMgr is a subclass that supports asynchronous I/O. While the interface to the connections is through the DtConnectionMgr, it is the DtAsyncConnectionMgr class that is instantiated by the RTI. The DtAsyncConnectionMgr class supports asynchronous or polling I/O based on RID file parameters. You can override the Connection Manager (from either DtConnectionMgr or DtAsyncConnectionMgr) using the DtConnectionMgr::setCreatorFunction() member function.

The DtAsyncConnectionMgr class handles all asynchronous I/O decisions by overriding member functions in DtConnectionManager and calling up to DtConnectionManager if asynchronous I/O is disabled.

For example, the tick member function calls processPendingMsg() and then processConnections() as follows:

RTI::Boolean DtConnectionMgr::tick()
{
   DtBoolean pendingResult = processPendingMsg();
   DtBoolean processResult = processConnections();

   return (processResult || pendingResult) ? RTI::RTI_TRUE :
      RTI::RTI_FALSE; 
}

The implementation of DtConnectionMgr::processConnections() processes the connections directly. DtAsyncConnectionMgr overrides the processConnections() member function. If asynchronous I/O is disabled, it calls up to DtConnectionManager:: processConnections() and returns. On the sending side, DtConnectionMgr::send(...) sends messages directly using the connections. The DtAsyncConnectionMgr overrides the send() method. The implementation of DtAsyncConnectionMgr::send(...) checks the asynchronous status. If asynchronous I/O is disabled, it calls up to DtConnectionMgr::send(...) and returns. If asynchronous is enabled, DtAsyncConnectionMgr::send(...) invokes DtAsyncConnectionMgr::queueMsg(...).

Several member functions in DtAsyncConnectionMgr encapsulate the signaling between federate and I/O thread.

// Signal the receive event
// Used by I/O thread to signal federate thread
virtual void signalReceiveEvent();

// Clear receive event
// Used by the federate thread to clear signal
virtual void clearReceiveEvent();

// Signal the send event
// Used by the federate thread to signal I/O thread
virtual void signalSendEvent();
// Clear send event
// Used by the I/O thread to clear signal
virtual void clearSendEvent();

The readFileDescriptor(RTI::TransportType transport) and readFileEvent(RTI::TransportType transport) member functions generically support checking the status of pending messages.

7.4 Changing the RTI's Communications Infrastructure

One thing you might like to do with the RTIspy API is change how the RTI communicates, for example you might want to have the RTI encrypt and decrypt the messages sent over the network.

To change the RTI's communications infrastructure, you need to change the way the connections are set up and configured. You can approach this with some or all of the following:

• Create new subclasses of the DtSocket class that DtRtiConnection uses
• Create new subclasses of DtRtiConnection
• Instantiate existing DtRtiConnections with different parameters.

Whatever approach you take, you will probably need to subclass DtConnectionMgr and override some of the functions responsible for creating connections.

7.4.1 Subclassing DtSocket

A DtRtiConnection uses a DtSocket to send and receive messages. Depending on which constructor you use, it can create its own or you can pass one in. If you subclass DtSocket, pass it to the DtRtiConnection. The RTIspy API uses the VR-Link DtSocket. To replace the MAK RTI's default network-based communication infrastructure by changing the behavior at the DtSocket level, you would probably want to:

1. Derive a new kind of DtSocket that implements sending and receiving by writing and reading using your alternative communication method. (For complete information about creating a DtSocket, please see VR-Link documentation.)
2. Derive a new kind of DtConnectionMgr that uses a single DtRtiConnection that is initialized by passing in an instance of your new type of socket.

7.4.2 The Role of DtRtiConnection

A DtRtiConnection is a wrapper around a DtSocket. A DtSocket just sends and receives messages. A DtRtiConnection knows what kind of messages they are. It has functions for creating and removing callbacks that can execute code in response to particular message types. It also has accessor and mutators for federation execution and federate handles and the federation execution name. It is probably not necessary to subclass DtRtiConnection, because the best way to change networking details is to subclass DtSocket.

7.4.3 Overriding Connection Manager Functions

If you want to completely reimplement the way we create connections, ignoring what the RID file says, and so on, you can ignore most of the functions in DtConnectionMgr, and just override init(), internalConn(), reliableConn(), and udpConn(). Subclass DtRtiConnection to create the modified connection that you want to use. Then have internalConn(), reliableConn(), and udpConn() return pointers to instances of your new connections.

If you want the Connection Manager to generally work like the default version, but need to make some small changes, please see the header files (or class documentation) for information about the various functions.

Overriding DtConnectionMgr::tick()

The tick() function calls readAndProcess(), which checks for new messages. The default is to call readAndProcess() on each connection. You probably will not want to change this behavior. However, if you do not want to check a particular connection every tick, you can override tick() to change the behavior.

7.4.4 Example

The following example demonstrates methods for changing how the RTI communicates, as discussed in this section.

1. Derive a new DtSocket that does sending and receiving using whatever communications mechanism you choose:

   class MySocket : public DtSocket
   {
   public:
   
      // Constructors
      ...
      // Override sendTo
      virtual int sendTo(caddr_t packet, size_t size,
         DtInetAddr addr = DtUSE_DEFAULT_IP_ADDR)
      {
         // Your implementation here. Return number of bytes sent, -1 for
         // failure.
      }
   
      // Override the two versions of DtRecv
      virtual int DtRecv(caddr_t* buffptr, int* status = NULL)
      {
         // Your implementation here. Set buffptr to point to a buffer that
         // contains the received data. Return number of bytes received, -1
         // for failure.
      }
      virtual int DtRecv(caddr_t buff, size_t size)
      {
         // Your implementation here. Copy received data into buff, up to
         // size bytes. Return number of bytes received, -1 for failure.
      }
   };

2. Derive a new DtConnectionMgr that uses a DtRtiConnection configured to use an instance of your new kind of DtSocket:

   class MyConnectMgr : public DtConnectionMgr
   {
   public:
   
      // Constructor
      MyConnectMgr(DtRIDParameters* params, DtFederateMgr* fedMgrPtr) :
         DtConnectionMgr(params, fedMgrPtr)
      {
      }
   
      // Destructor
      ~MyConnectMgr
      {
         // Since we create the socket and connection (in init), we should
         // delete them. Set myUdpConn, myTcpConn, and myInternalConn to NULL
         // so that the base destructor will not also try to delete them. In
         // this example, all three pointers point to the same connection, so
         // only delete it once. 
         delete myUdpConn;
         myUdpConn = myReliableConn = myInternalConn = NULL;
      }
   
      virtual void init(DtRIDParameters* params)
      {
         // Instantiate your derived socket, and a DtRtiConnection that will
         // use it
         DtSocket* sock = new MySocket(...);
         DtRtiConnection* conn = new DtRtiConnection(sock);
   
         // In this example, we set up myUdpConn, myReliableConn, and
         // myInternalConn to point to this one connection. Alternatively,
         // you might not want to use a single physical connection for all
         // three logical connections.
         myUdpConn = myReliableConn = myInternalConn = conn;
      }
   
      // Creator function, which we will register with the RTI
      static DtConnectionMgr* create(DtRIDParameters* params, 
         DtFederateMgr* fedMgrPtr)
      {
         return new MyConnectMgr(params, fedMgrPtr);
      }
   };

3. Tell the RTI to use your derived connection manager, within SetPluginCreators():

   void SetPluginCreators()
   {
      DtConnectionMgr::setCreatorFunction(MyConnectMgr::create);
   }

7.5 Messages

Messages in the RTI are represented by instances of subclasses of DtRtiMessage (rtiMsg.h). Different subclasses represent different kinds of messages that an LRC needs to send to other LRCs or to the rtiexec. Examples include DtJoinMsg (joinMsg.h), which is sent by a federate as a result of a call to joinFederateExecution(), and DtUpdateRoMsg (updateRoMsg.h), which contains receive order attribute updates. For a full list of message kinds, please see rtiMsg.h.

You might want to use the RTIspy API to change the wire format for messages. This typically involves subclassing one or more of the DtRtiMsgs and changing the way the message is represented on the network. For example, to change the way attribute updates are encoded, subclass from DtUpdateRoMsg, DtUpdateTsoMsg, or both.

You can also create brand new types of messages, but these will only be useful if you are creating custom managers and are writing code to send and receive the new messages.

The various DtRtiMsg subclasses have accessor functions to set and inspect various fields. All messages share common header information. Accessors for header fields are inherited from the base DtRtiMsg. DtRtiMsg inherits from DtBaseMsg, which is taken from the vlutil library in the VR-Link API. When a DtRtiMsg is sent through a DtRtiConnection, the connection calls netRep() on the message to get a buffer that contains exactly the bytes that need to be communicated to other LRCs or the rtiexec. It then passes that buffer to its DtSocket for sending. On the receiving side, when a connection gets a packet from its DtSocket, it constructs the appropriate kind of DtRtiMsg by passing that network representation buffer to the message's constructor.

When deriving your own kinds of DtRtiMsgs, you could just ignore the way we normally implement the accessors, netRep(), and the from-network-representation constructors, and provide brand new implementations for these functions. However, we suggest taking advantage of the infrastructure that the DtRtiMsg class provides for managing the buffer used to store the network representation, managing the message headers, and so on. The recommended method for subclassing DtRtiMsg very closely parallels the method of deriving new kinds of DtPdus in VR-Link. For examples, please see the VR-Link documentation.

Once you have created new subclasses of DtRtiMsg, you must instruct the RTI to use them instead of the default versions. Do this by properly configuring the RTI's message factory. A message factory, represented by the DtRtiMsgFactory class (rtiMsgFact.h), is a table that maps message types to creator functions that are used to instantiate the appropriate subclass of DtRtiMsg. When the RTI needs to instantiate a particular kind of message, it looks up the message kind in the DtRtiMsgFactory, finds the appropriate creator function, and uses it to create the right type of message. The DtRtiMsgFactory is obtained from the Connection Manager, using its msgFactory() member.

To configure the message factory so that it will contain your creator functions, use its addSendCreator(), and addReceiveCreator() member, passing the message kind, along with a creator function to be used for outgoing or incoming messages. The following example assumes you are creating a subclass of DtUpdateMsg called MyUpdateRoMsg:

class MyUpdateRoMsg : public DtUpdateRoMsg
{
public:

   // Creator function for sending:
   static DtMsg* create(DtBufferPtr buffer = DtUSE_INTERNAL_BUFFER)
   {
      return new MyUpdateRoMsg(buffer);
   }

   // Creator function for receiving
   static DtMsg* create(const DtNetMsgHeader* initial,
      DtBufferPtr buffer = DtUSE_INTERNAL_BUFFER)
   {
      return new MyUpdateRoMsg(initial, buffer);
   }

   ...
   // Constructors, and overridden member functions
   ...
};

void InitPlugin(DtRtiAmbassadorImplementor* implementor)
{
   DtRtiMsgFactory* fact = implementor->connectMgr()->msgFactory();
   fact->addSendCreator(DtUpdateRoMsgKind, MyUpdateRoMsg::create);
   fact->addReceiveCreator(DtUpdateRoMsgKind, MyUpdateRoMsg::create);
}

[<< Subclassing DtRtiAmbassadorImplementor] [Home] [Top of Page] [Responding to Dropped Federates >>]