Serializing RTI DDS traffic

Hello Arthur,

This sounds very interesting!

I think the best way to accomplish what you want is to use the RTI Connext DDS Pluggable Transport API.

The  Pluggable Transport API sits below the DDS protocol layer and receives a fully serialized RTPS message along with the destination to where it should be sent.  Connext DDS uses this API and bundles implementations that use UDP/IP, Shared Memory, TCP/IP, TLS, etc.   We have also used this API to develop specialized transports for some customers, things like Infiniband, PCI/Express backplane, low-bandwidth transports that do compression and protocol reduction, etc.  

The nice thing about this approach is that you do not really need to understand/interpret the RTPS messages (although you could) all you need is to provide the mans for a bunch of bytes to be delivered from point A to point B.   The transport plugin implementation provides the definition of what an "address" is and what a "port" is and maps it into whatever it makes sense for that transport.

The APIs are all bundled with the Connext DDS distribution and are mentioned in the on-line docs. However but you do not have any examples or the full interface documentation in the standard distribution so it would be hard, if not impossible to figure out from this.

I uploaded a ZIP file with the documentation to the File Exchange. You can access it at this link: File Exchange > Additional Documentation > pluggable_transport_doc.zip.

Once you unzip the file you will get a directory. Open the "index.html" file there and you will be able to browse the documentation.

One reason we do not bundle this with the standard Connext DDS download is that creating a pluggable transport is a bit involved and the documentation itself is probably not enough for your to figure out what to do. I think the thing that would really help you is to have a good example on how this API is used.  Unfortunately I did not see anything that was customer-ready.  I will try to create one for you and will follow up on this thread as soon as it is ready. Hopefully by the end of the week.

Gerardo

Hi Arthur,

The documents you found using the search engine are indeed the documents Gerardo uploaded.

Fernando.

Hi Arthur,

Not sure why you are getting the "invalid error". I tried it from a different account and it worked OK. Are you clicking on the "Download" that appears in the menubar (see below)?

To answer your question, about "grabbing" the RTPS objects.  I think the answer is yes.

Once you install a Transport Plugin it will be used in addition (or instead) of the other transports you have installed. So for example, if you intstall your DTN transport and disable UDP  Connext DDS will use only the DTN. You could also configure it to use both, or to use either depending on the destination.  Another thing you could do it install the DTN transport plugin as a "decorator" pattern over the regular UDP transport, so you can intercept each RTPS messages and decide whether to dispatch them to the decorated UDP transport or handle it yourself, or pre-process it and then handle it to the UDP transport (the latter is what our "compression" transport does).

Gerardo

Hello Arthur,

Sorry for the delay.  I was not finding much extra time to work on this, but finally managed to get what I think is a simple but realistic example of creating a custom transport. You can browse and downoad the example at this URL:  http://community.rti.com/examples/creating-custom-transport. Let me know what you think.

Regarding your second question. Are you asking how to interpret RTPS messages, or the application data thay encapsulate?

RTPS messages are things like HeartBeat, AckNack, Data, DataFrag, InfoTimestap, etc.  The format is described in the standard DDS Interoperability Protocol, Real-Time Publish-Subscribe Wire Protocol (DDS-RTPS). You can find some code that interprets the RTPS messages in the open source Wireshark protocol analyzer.  The Wireshark RTPS dissector is actively maintained by RTI.

Some of the RTPS messages contain serialized application data (Data, DataFrag).  If you want to parse looking at the bytes on the wire you first need to parse the RTPS submessage to get to the actual serialized payload. And then you need to use the appropriate "deserialize" function for that type. For this last part you can use the DynamicData API that we talked about in the thread http://community.rti.com/forum-topic/dynamicdata-api-usage-receiving-arbitrary-data-topic.

I am not sure if this is what you were asking... If I did not understand correctly please elaborate and I will try again.

Regards,

Gerardo

Hi Arthur,

Yes, in principle it would be possible to have the transport plugin traverse the RTPS message to find the Data submessage and the SerializedData submessage and then interpret the CDR stream to find the fields you are looking for...  However there is a couple of things that will make the setup tricky.  For example the RTPS Data submessage does not contain the Topic name. Only the GUID of the DataWriter that wrote the message. You would need to determine know that association at run-time and pass that informatpion to the transport plugin, or else you could have the transport plugin also interpret the Data messages sent by the PublicationBuiltinTopic and find from those the mapping of DataWriter GUID and Topic name... This requires getting up to speed on the RTPS protocol... The Wireshark source code for the RTPS dissector could help but it still sounds like a lot of work...

You said the criteria for filtering is based on the value of fields in the data. In that case why can't you use the regular ContentBasedFilter facility? These filters are applied on the writer side so as long as yoiu can set up the filtes such that only one filter passes for each sample you should be able to get the same effect. In fact even if the builtin SQL filters cannot do this, you could implement your own filter functions (via a plugin) and load it dynamically. Likely this would give you the flexibility you need because the filter decision can be stateful and you can enforce that only one receiver gets each sample.

Regarding whether the transport plugin must be in C...  Practically yes. You can implement it in C or C++, the API itself is in C, but you could have the guts of it in C++ and then wrap the C++ methods with the C functions defined by the Transport plugin interfarce.

Gerardo

Hi Arthur,

The fact that you are using the Java API to Connext DDS is not an issue. The transport plugins are loaded by the RTI DDS core, which is in C. And the Java API sits on top of all of it so it is not aware of the implementation technology for the transport.

As far as implementing the Transport Plugin in something other than C/C++...  To me it would appear the easiest thing would be to find a DTN implementation in C/C++ :)   It seems there are some out there like http://www.dtnrg.org/wiki/Code doing it this way would likely get you better performance and also simplify debugging as you would not have to deal with the JNI layer... But I understood you are using your own DTN implementation so this is not a real option...

I think it is certainly possible to implement the transport plugin in Java. You would have to implement the plugin the core expects in C and use JNI to call the Java classes.  We have done similar things in our Routing Service product which has an Adaptor API natively in C but we also support users implementing the adaptor in Java. To do this we created the JNI code to load the Java classes and invoke them. I found this web page http://www.codeproject.com/Articles/22881/How-to-Call-Java-Functions-from-C-Using-JNI with pointers on how to do it

Regards,

Gerardo

Hi Arthur,

The RTI_INT32 type is defined in osapi_type.h  but this header is already included from osapi_process_h so you should not include it directly because you would be masking the real problem...

The issue you are having is that you are missing some #defines that are required to select the right platform and such...

How are you building from eclipse? Did you create your own makefile that calls some Windows compiler? Is that VisualStudio or something else? or were you using eclipse just to browse and it is the indexer that is failing to resolve the RTI_WIN32?

If you are using your own makefile/visual studio project then the easiest thing I would suggest is to run rtiddsgen selection the platform you are using or something close. For example i86Win32VS2008 or x64Win64VS2008 for 64 bits. Then open that Microsoft VisualStudio project (MSVC) and see the defines it has for that platform and copy those to your eclipse project. I have not tried this myself so your mileage may vary...

For example this is what appears on my Windows XP for 32 bits (this is the one VM I have handy):  WIN32;RTI_WIN32;NDEBUG;_CONSOLE

So for sure in this platform you would need the RTI_WIN32 (to get our header files correctly). I think the NDEBUG;_CONSOLE is specific to the MSVC compiler but I am unsure on the WIN32. It may be we also need it or it could be required just by the MSVC compiler...

As to the buffers for sending/receiving...  Can you be a bit more specific on which buffers are you referring to? Transport buffers you would need to define yourself in the plugin implementation. The middleware buffers are configured via QoS.

Gerardo

Hello Arthur,

Did you uncheck the "Use default build command" checkbox on the Eclipse C/C++ build configuration window? If you leave that checked as in the screenshot below then Eclipse will not let you edit the Build command. As soon as you uncheck that option the build command should became editable.

As far as the indexer I was able to get the Eclipse Indexer to resolve all references as soon as I added both ${NDDSHOME}/include and ${NDDSHOME}/include/ndds to the "Includes" tab in "Paths and Symbols" and also defined RTI_WIN32 on the "# Symbols" tab also in "Paths and Symbols". See below.

That said, I would really recommend you use Microsoft Visual Studio for C/C++ development on Windows because this is the toolchain we support. In fact rtiddsgen will create the Visual Studio projects which are easy to modify to build the transport DLLs. You can use the Express Edition of Visual Studio which is free...

If I understood correctly your question regarding the buffers, you are referring to the buffer_in parameter to the NDDS_Transport_Send_Fnc that your plugin must implement. Is that right?

In that case the buffer_in represents memory that is allocated and managed by the core and passed to your Plugin function.  It is your Plugin responsibility to send the data contained in those buffers but you do not need to create or release that memory...  

The DDS core will make sure that on each call to the Transport plugin  NDDS_Transport_Send_Fnc operation the total number of buffers in buffer_indoes not exceed gather_send_buffer_count_max and the total number of bytes contained accross all buffers in buffer_indoes not exceed the message_size_max. Both of these are properties that the transport Plugin implementation controls.  They are part of the NDDS_Transport_Property_t which is configured when the Plugin is created. In the FileTransport example I used the values NDDS_TRANSPORT_FILE_GATHER_SEND_BUFFER_COUNT_MAX_DEFAULT and NDDS_TRANSPORT_FILE_MESSAGE_SIZE_MAX_DEFAULT which are defined in FileTransport.h

I am not sure if this was really your question. If not let me know and I will try again...

Regards,

Gerardo

Hello,

Oh I see. It is odd that Eclipse does not let you modify that setting. I have not seen that before... Is it that it thinks the project is read only? Anyway it seems like using Visual Studio is an option for you so I think that is a better way to proceed...

Quick note: I am not sure what you mean by "replaced all RTI_INT32 with RTI_WIN32". Was that a typo?  RTI_INT32 should not be replaced with RTI_WIN32...

As to how long it takes to fill the  "buffer_in" for a gather send?  I guess it depends on the platform and the QoS settings.  With the out-of-the-box configuration:

• ASYNCHRONOUS_PUBLISHER QoS set to its default value which makes writes synchronous
• BATCH QoS set to its default value which makes the middleware send individual samples without tryign to group then=m
• PUBLISH_MODE set to its default value which allows data to be sent without a FlowController or output shaper.

Then the DDS middleware core will fill in the gather bufers and call send as fast as it can as a result of the application calling DataWriter::write().  How long that takes obviously depends on the size of the data and the speed of the computer.  The data has to be serialized, copied into the middleware cache, appended with a timestamp and sequence number, the RTPS packet must be formed, etc.  For small samples is modern Intel CPUs the total time from calling DataWriter::write() to having the middleware call the transport NDDS_Transport_Send_Fnc should be 5 to 20 microseconds. That would be my guess.

Gerardo

Hi Arthur,

Regarding the mapping from IPv6 address/port to filename I will first describe the transport-plugin model a little before getting to the specific answer to your question...

Each transport class (identified by the attribute classid within the NDDS_Transport_ClassId_t structure) identifies a separate Transport Plugin. The classid if of type NDDS_Transport_ClassId_t which is just a typedef for a 32-bit integer.  There is a number of pre-defined values used by existing RTI transports, such as 1 for UDP/IP v2, 2 for Shared Memory, etc. These are all listed in the documentation of NDDS_Transport_ClassId_t.  As a side not there you will see the definition of NDDS_TRANSPORT_CLASSID_RESERVED_RANGE (1000) which marks the reserved range classid. User-defined transports should have a value greater than this number. Which I see my FileTransport example violates so I will update the example to correct this.

So the first think when defining a new TransportPlugin is pick the classid.   This must happen as part of the TransportPlugin creation process because the function that you register with DDS to create the Transport must return a  NDDS_Transport_PluginImpl (note that NDDS_Transport_Plugin is just a typedef of struct NDDS_Transport_PluginImpl). So basically as part of the creation of your transport you must fill the  NDDS_Transport_PluginImpl which contains the NDDS_Transport_Property_t with classid and other configuration settings as well as all the function pointers that do the work of opening send/receive resources, sending data and receiving data.

Each NDDS_Transport_ClassId_t creates its own namespace for addresses. So the full an address of an endpoint is the triplet (NDDS_Transport_ClassId_t, DDS_Transport_Address_t, NDDS_Transport_Port_t).  This triplet maps exactly to the DDS_Locator_t which is what the DDS-RTPS standard specifies should be communicated on the wire to describe a DDS-RTPS endpoint.

The RTI DDS Core automatically dispatches transport related calls to the right TransportPlugin based on the NDDS_Transport_ClassId_t, so each TransportPlugin will only see the (DDS_Transport_Address_t, NDDS_Transport_Port_t) part. You can see these in the signatures of transport-plugin functions such as NDDS_Transport_Create_RecvResource_Fcn_rrEA and NDDS_Transport_Create_SendResource_Fcn_srEA.

The consequence of the fact that RTI DDS dispatches to a transport-plugin based on the NDDS_Transport_ClassId_t is that each NDDS_Transport_ClassId_t creates a separate 'namespace' for the (DDS_Transport_Address_t, NDDS_Transport_Port_t). The  DDS_Transport_Address_t is a 16-byte address (i.e. an IPv6 address) and the NDDS_Transport_Port_t is a 32-bit integer (i.e. a UDPv6 port). Therefore you can reuse the same "IPv6" address and port for each transport plugin and define them as you see fit for each transport.

So a transport like IPv6 can use its full range of legal addresses without restrictions. Other transport plugins need to map whichever they use as their internal addressing mechanism to a 16-byte address and32-bit port.  The mapping is private to the transport plugin and needs to be understood only by that transport plugin.

One final twist on this is that as part of the transport definition provided by the NDDS_Transport_PluginImpl is a property called address_bit_count which defines how many addressing bits the transport really uses. That is the address is 16 Bytes (i.e. 128 bits) but a Transport Plugin may use fewer bits and it indicates it with the setting of the address_bit_count field. 

In the FileExample you can see how it is configured in the definition of NDDS_TRANSPORT_FILE_PROPERTY_DEFAULT that appears in FileTransport.h, which I am copying below:

 #define NDDS_TRANSPORT_FILE_PROPERTY_DEFAULT { \
    { NDDS_TRANSPORT_CLASSID_FILE, \
      NDDS_TRANSPORT_FILE_ADDRESS_BIT_COUNT, \
      NDDS_TRANSPORT_FILE_PROPERTIES_BITMAP_DEFAULT, \
      NDDS_TRANSPORT_FILE_GATHER_SEND_BUFFER_COUNT_MAX_DEFAULT, \
      NDDS_TRANSPORT_FILE_MESSAGE_SIZE_MAX_DEFAULT, \
      NULL, 0, /* allow_interfaces_list */ \
      NULL, 0, /* deny_interfaces_list */ \
      NULL, 0, /* allow_multicast_interfaces_list */ \
      NULL, 0, /* deny_multicast_interfaces_list */ \
    }, \
    NDDS_TRANSPORT_FILE_RECEIVED_MESSAGE_COUNT_MAX_DEFAULT, \
    NDDS_TRANSPORT_FILE_RECEIVE_BUFFER_SIZE_DEFAULT,    \
    NDDS_TRANSPORT_FILE_TRACE_LEVEL_DEFAULT, \
    NDDS_TRANSPORT_FILE_ADDRESS_DEFAULT, \
    NDDS_TRANSPORT_FILE_DIRECTORY_ROOT_DEFAULT \
} 

If you look in that file you will see I defined NDDS_TRANSPORT_FILE_ADDRESS_BIT_COUNT to be 32. So I am only really considering the last 4-Bytes (32 bits) as the part of the 16 Byte address that the FileTransport plugin cares about.

Now to your orginal question on how the FileTransport maps IPv6 addresses (and ports)  to filenames... From the description above the mapping is "arbitrary" the only thing that matters is that the FileTransport does it consistently on the sending and receiving sides. Since this is a simple example I used a very simple approach. There is some documentation in the comments at the beginning of  FileTransport.c. Basically all the files are placed under a common "base directory", which defaults to the compiled-in variable NDDS_TRANSPORT_FILE_DIRECTORY_ROOT_DEFAULT ( /tmp/dds/FileTransport) but can also be configured setting the property " dds.transport.FILE.myPlugin.directory" in the USER_QOS_PROFILE.xml provided in the example.  Note that the use of "myPlugin" in the property name is specific to the fact that we used that name when we configured the plugins to load by setting the value of the property " dds.transport.load_plugins".

So the base directory is configured (or defaulted) at the  time the FileTransport is loaded to RTI DDS. Given that the mapping of (address, port) to filename is done by the function NDDS_Transport_FILE_getfilename_for_address_and_port in FileTransport.c which basically uses the last 4 bytes of the address (since we defined the TransportPlugin to only pay attention to the last 32 bits) and the port to create a filename relative to the base directory. The this is the code within the function that does the mapping:

    sprintf(fname_out, "%s/%d.%d.%d.%d/%d", NDDS_Transport_FILE_getdirname(me),
            addrBytes[12], addrBytes[13], addrBytes[14], addrBytes[15],
            port_in);

The result is that an IP address such as 10.1.37.8 and port 1234 would be mapped to the file: /tmp/dds/FileTransport/10.1.37.8/1234.

Finally the configuration of the addresses... This is the responsibility of the transport plugin. The DDS core calls the function NDDS_Transport_Get_Receive_Interfaces_Fcn_cEA and uses whichever addresses that function returns.  In the FileExample I made it manual so that each FileExample plugin instantiation is configured with its own address. This is done with the property dds.transport.FILE.myPlugin.address. I  could have done it differently, for example, come up with the address automatically hashing the computer hostId and processId. But this would have just complicated the code and this was intended to be a simple example... The idea is that in a realistic case the "address" of each TransportPlugin is something native and natural to the TransportPlugin itself...

I hope this clarifies things a bit. I realize that this interface is a bit complex, but it is needed to support the many different transports out there. We have used it for things as diverse as shared memory, DMA type transports, Infiniband, serial, connectionless and datagram based like UPD, connection-oriented and stream-oriented like TCP, TLS, NAT Traversing transports and several others...

Gerardo