stricter warning checks and the corresponding fixes
Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson
eboasson
parent
87e4780446
commit
b7487b18a6
101 changed files with 1756 additions and 1615 deletions
|
|
@ -10,6 +10,8 @@ int main (int argc, char ** argv)
|
|||
dds_entity_t writer;
|
||||
dds_return_t ret;
|
||||
HelloWorldData_Msg msg;
|
||||
(void)argc;
|
||||
(void)argv;
|
||||
|
||||
/* Create a Participant. */
|
||||
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
|
||||
|
|
|
|||
|
|
@ -17,6 +17,8 @@ int main (int argc, char ** argv)
|
|||
dds_sample_info_t infos[MAX_SAMPLES];
|
||||
dds_return_t ret;
|
||||
dds_qos_t *qos;
|
||||
(void)argc;
|
||||
(void)argv;
|
||||
|
||||
/* Create a Participant. */
|
||||
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
|
||||
|
|
|
|||
|
|
@ -65,7 +65,7 @@ static ExampleTimeStats *exampleAddTimingToTimeStats
|
|||
{
|
||||
stats->values[stats->valuesSize++] = timing;
|
||||
}
|
||||
stats->average = (stats->count * stats->average + timing) / (stats->count + 1);
|
||||
stats->average = ((double)stats->count * stats->average + (double)timing) / (double)(stats->count + 1);
|
||||
stats->min = (stats->count == 0 || timing < stats->min) ? timing : stats->min;
|
||||
stats->max = (stats->count == 0 || timing > stats->max) ? timing : stats->max;
|
||||
stats->count++;
|
||||
|
|
@ -117,8 +117,9 @@ static bool CtrlHandler (DWORD fdwCtrlType)
|
|||
return true; //Don't let other handlers handle this key
|
||||
}
|
||||
#else
|
||||
static void CtrlHandler (int fdwCtrlType)
|
||||
static void CtrlHandler (int sig)
|
||||
{
|
||||
(void)sig;
|
||||
dds_waitset_set_trigger (waitSet, true);
|
||||
}
|
||||
#endif
|
||||
|
|
@ -149,14 +150,14 @@ static dds_time_t elapsed = 0;
|
|||
|
||||
static bool warmUp = true;
|
||||
|
||||
static void data_available(dds_entity_t reader, void *arg)
|
||||
static void data_available(dds_entity_t rd, void *arg)
|
||||
{
|
||||
dds_time_t difference = 0;
|
||||
int status;
|
||||
(void)arg;
|
||||
/* Take sample and check that it is valid */
|
||||
preTakeTime = dds_time ();
|
||||
status = dds_take (reader, samples, info, MAX_SAMPLES, MAX_SAMPLES);
|
||||
status = dds_take (rd, samples, info, MAX_SAMPLES, MAX_SAMPLES);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
postTakeTime = dds_time ();
|
||||
|
||||
|
|
@ -218,8 +219,8 @@ static void usage(void)
|
|||
|
||||
int main (int argc, char *argv[])
|
||||
{
|
||||
unsigned long payloadSize = 0;
|
||||
unsigned long long numSamples = 0;
|
||||
uint32_t payloadSize = 0;
|
||||
uint64_t numSamples = 0;
|
||||
bool invalidargs = false;
|
||||
dds_time_t timeOut = 0;
|
||||
dds_time_t time;
|
||||
|
|
@ -271,9 +272,11 @@ int main (int argc, char *argv[])
|
|||
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
|
||||
DDS_ERR_CHECK (participant, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
listener = dds_listener_create(NULL);
|
||||
dds_lset_data_available(listener, data_available);
|
||||
|
||||
if (use_listener)
|
||||
{
|
||||
listener = dds_listener_create(NULL);
|
||||
dds_lset_data_available(listener, data_available);
|
||||
}
|
||||
prepare_dds(&writer, &reader, &readCond, listener);
|
||||
|
||||
setvbuf(stdout, NULL, _IONBF, 0);
|
||||
|
|
@ -300,7 +303,7 @@ int main (int argc, char *argv[])
|
|||
}
|
||||
if (argc - argidx >= 1)
|
||||
{
|
||||
payloadSize = atol (argv[argidx]);
|
||||
payloadSize = (uint32_t) atol (argv[argidx]);
|
||||
|
||||
if (payloadSize > 100 * 1048576)
|
||||
{
|
||||
|
|
@ -309,7 +312,7 @@ int main (int argc, char *argv[])
|
|||
}
|
||||
if (argc - argidx >= 2)
|
||||
{
|
||||
numSamples = atol (argv[argidx+1]);
|
||||
numSamples = (uint64_t) atol (argv[argidx+1]);
|
||||
}
|
||||
if (argc - argidx >= 3)
|
||||
{
|
||||
|
|
@ -317,7 +320,7 @@ int main (int argc, char *argv[])
|
|||
}
|
||||
if (invalidargs || (argc - argidx == 1 && (strcmp (argv[argidx], "-h") == 0 || strcmp (argv[argidx], "--help") == 0)))
|
||||
usage();
|
||||
printf ("# payloadSize: %lu | numSamples: %llu | timeOut: %" PRIi64 "\n\n", payloadSize, numSamples, timeOut);
|
||||
printf ("# payloadSize: %" PRIu32 " | numSamples: %" PRIu64 " | timeOut: %" PRIi64 "\n\n", payloadSize, numSamples, timeOut);
|
||||
|
||||
pub_data.payload._length = payloadSize;
|
||||
pub_data.payload._buffer = payloadSize ? dds_alloc (payloadSize) : NULL;
|
||||
|
|
@ -418,7 +421,7 @@ done:
|
|||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
static dds_entity_t prepare_dds(dds_entity_t *writer, dds_entity_t *reader, dds_entity_t *readCond, dds_listener_t *listener)
|
||||
static dds_entity_t prepare_dds(dds_entity_t *wr, dds_entity_t *rd, dds_entity_t *rdcond, dds_listener_t *listener)
|
||||
{
|
||||
dds_return_t status;
|
||||
dds_entity_t topic;
|
||||
|
|
@ -448,8 +451,8 @@ static dds_entity_t prepare_dds(dds_entity_t *writer, dds_entity_t *reader, dds_
|
|||
dwQos = dds_qos_create ();
|
||||
dds_qset_reliability (dwQos, DDS_RELIABILITY_RELIABLE, DDS_SECS (10));
|
||||
dds_qset_writer_data_lifecycle (dwQos, false);
|
||||
*writer = dds_create_writer (publisher, topic, dwQos, NULL);
|
||||
DDS_ERR_CHECK (*writer, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
*wr = dds_create_writer (publisher, topic, dwQos, NULL);
|
||||
DDS_ERR_CHECK (*wr, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (dwQos);
|
||||
|
||||
/* A DDS_Subscriber is created on the domain participant. */
|
||||
|
|
@ -463,17 +466,17 @@ static dds_entity_t prepare_dds(dds_entity_t *writer, dds_entity_t *reader, dds_
|
|||
/* A DDS_DataReader is created on the Subscriber & Topic with a modified QoS. */
|
||||
drQos = dds_qos_create ();
|
||||
dds_qset_reliability (drQos, DDS_RELIABILITY_RELIABLE, DDS_SECS(10));
|
||||
*reader = dds_create_reader (subscriber, topic, drQos, listener);
|
||||
DDS_ERR_CHECK (*reader, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
*rd = dds_create_reader (subscriber, topic, drQos, listener);
|
||||
DDS_ERR_CHECK (*rd, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (drQos);
|
||||
|
||||
waitSet = dds_create_waitset (participant);
|
||||
if (listener == NULL) {
|
||||
*readCond = dds_create_readcondition (*reader, DDS_ANY_STATE);
|
||||
status = dds_waitset_attach (waitSet, *readCond, *reader);
|
||||
*rdcond = dds_create_readcondition (*rd, DDS_ANY_STATE);
|
||||
status = dds_waitset_attach (waitSet, *rdcond, *rd);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
} else {
|
||||
*readCond = 0;
|
||||
*rdcond = 0;
|
||||
}
|
||||
status = dds_waitset_attach (waitSet, waitSet, waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
|
@ -481,19 +484,19 @@ static dds_entity_t prepare_dds(dds_entity_t *writer, dds_entity_t *reader, dds_
|
|||
return participant;
|
||||
}
|
||||
|
||||
static void finalize_dds(dds_entity_t participant, dds_entity_t reader, dds_entity_t readCond)
|
||||
static void finalize_dds(dds_entity_t ppant, dds_entity_t rd, dds_entity_t rdcond)
|
||||
{
|
||||
dds_return_t status;
|
||||
|
||||
/* Disable callbacks */
|
||||
dds_set_enabled_status (reader, 0);
|
||||
dds_set_enabled_status (rd, 0);
|
||||
|
||||
(void) dds_waitset_detach (waitSet, readCond);
|
||||
(void) dds_waitset_detach (waitSet, rdcond);
|
||||
status = dds_waitset_detach (waitSet, waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
(void) dds_delete (readCond);
|
||||
(void) dds_delete (rdcond);
|
||||
status = dds_delete (waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_delete (participant);
|
||||
status = dds_delete (ppant);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,8 +21,9 @@ static bool CtrlHandler (DWORD fdwCtrlType)
|
|||
return true; //Don't let other handlers handle this key
|
||||
}
|
||||
#else
|
||||
static void CtrlHandler (int fdwCtrlType)
|
||||
static void CtrlHandler (int sig)
|
||||
{
|
||||
(void)sig;
|
||||
dds_waitset_set_trigger (waitSet, true);
|
||||
}
|
||||
#endif
|
||||
|
|
|
|||
|
|
@ -1,308 +1,309 @@
|
|||
#include "ddsc/dds.h"
|
||||
#include "Throughput.h"
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <signal.h>
|
||||
|
||||
/*
|
||||
* The Throughput example measures data throughput in bytes per second. The publisher
|
||||
* allows you to specify a payload size in bytes as well as allowing you to specify
|
||||
* whether to send data in bursts. The publisher will continue to send data forever
|
||||
* unless a time out is specified. The subscriber will receive data and output the
|
||||
* total amount received and the data rate in bytes per second. It will also indicate
|
||||
* if any samples were received out of order. A maximum number of cycles can be
|
||||
* specified and once this has been reached the subscriber will terminate and output
|
||||
* totals and averages.
|
||||
*/
|
||||
|
||||
#define MAX_SAMPLES 100
|
||||
|
||||
static bool done = false;
|
||||
|
||||
/* Forward declarations */
|
||||
static dds_return_t wait_for_reader(dds_entity_t writer, dds_entity_t participant);
|
||||
static void start_writing(dds_entity_t writer, ThroughputModule_DataType *sample,
|
||||
unsigned int burstInterval, unsigned int burstSize, unsigned int timeOut);
|
||||
static int parse_args(int argc, char **argv, uint32_t *payloadSize, unsigned int *burstInterval,
|
||||
unsigned int *burstSize, unsigned int *timeOut, char **partitionName);
|
||||
static dds_entity_t prepare_dds(dds_entity_t *writer, const char *partitionName);
|
||||
static void finalize_dds(dds_entity_t participant, dds_entity_t writer, ThroughputModule_DataType sample);
|
||||
|
||||
/* Functions to handle Ctrl-C presses. */
|
||||
#ifdef _WIN32
|
||||
#include <Windows.h>
|
||||
static int CtrlHandler (DWORD fdwCtrlType)
|
||||
{
|
||||
done = true;
|
||||
return true; /* Don't let other handlers handle this key */
|
||||
}
|
||||
#else
|
||||
struct sigaction oldAction;
|
||||
static void CtrlHandler (int fdwCtrlType)
|
||||
{
|
||||
done = true;
|
||||
}
|
||||
#endif
|
||||
|
||||
int main (int argc, char **argv)
|
||||
{
|
||||
uint32_t payloadSize = 8192;
|
||||
unsigned int burstInterval = 0;
|
||||
unsigned int burstSize = 1;
|
||||
unsigned int timeOut = 0;
|
||||
char * partitionName = "Throughput example";
|
||||
dds_entity_t participant;
|
||||
dds_entity_t writer;
|
||||
ThroughputModule_DataType sample;
|
||||
|
||||
/* Register handler for Ctrl-C */
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler ((PHANDLER_ROUTINE) CtrlHandler, true);
|
||||
#else
|
||||
struct sigaction sat;
|
||||
sat.sa_handler = CtrlHandler;
|
||||
sigemptyset (&sat.sa_mask);
|
||||
sat.sa_flags = 0;
|
||||
sigaction (SIGINT, &sat, &oldAction);
|
||||
#endif
|
||||
|
||||
if (parse_args(argc, argv, &payloadSize, &burstInterval, &burstSize, &timeOut, &partitionName) == EXIT_FAILURE) {
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
participant = prepare_dds(&writer, partitionName);
|
||||
|
||||
/* Wait until have a reader */
|
||||
if (wait_for_reader(writer, participant) == 0) {
|
||||
printf ("=== [Publisher] Did not discover a reader.\n");
|
||||
DDS_ERR_CHECK (dds_delete (participant), DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
/* Fill the sample payload with data */
|
||||
sample.count = 0;
|
||||
sample.payload._buffer = dds_alloc (payloadSize);
|
||||
sample.payload._length = payloadSize;
|
||||
sample.payload._release = true;
|
||||
for (uint32_t i = 0; i < payloadSize; i++) {
|
||||
sample.payload._buffer[i] = 'a';
|
||||
}
|
||||
|
||||
/* Register the sample instance and write samples repeatedly or until time out */
|
||||
start_writing(writer, &sample, burstInterval, burstSize, timeOut);
|
||||
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler (0, false);
|
||||
#else
|
||||
sigaction (SIGINT, &oldAction, 0);
|
||||
#endif
|
||||
|
||||
/* Cleanup */
|
||||
finalize_dds(participant, writer, sample);
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
static int parse_args(
|
||||
int argc,
|
||||
char **argv,
|
||||
uint32_t *payloadSize,
|
||||
unsigned int *burstInterval,
|
||||
unsigned int *burstSize,
|
||||
unsigned int *timeOut,
|
||||
char **partitionName)
|
||||
{
|
||||
int result = EXIT_SUCCESS;
|
||||
/*
|
||||
* Get the program parameters
|
||||
* Parameters: publisher [payloadSize] [burstInterval] [burstSize] [timeOut] [partitionName]
|
||||
*/
|
||||
if (argc == 2 && (strcmp (argv[1], "-h") == 0 || strcmp (argv[1], "--help") == 0))
|
||||
{
|
||||
printf ("Usage (parameters must be supplied in order):\n");
|
||||
printf ("./publisher [payloadSize (bytes)] [burstInterval (ms)] [burstSize (samples)] [timeOut (seconds)] [partitionName]\n");
|
||||
printf ("Defaults:\n");
|
||||
printf ("./publisher 8192 0 1 0 \"Throughput example\"\n");
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
if (argc > 1)
|
||||
{
|
||||
*payloadSize = atoi (argv[1]); /* The size of the payload in bytes */
|
||||
}
|
||||
if (argc > 2)
|
||||
{
|
||||
*burstInterval = atoi (argv[2]); /* The time interval between each burst in ms */
|
||||
}
|
||||
if (argc > 3)
|
||||
{
|
||||
*burstSize = atoi (argv[3]); /* The number of samples to send each burst */
|
||||
}
|
||||
if (argc > 4)
|
||||
{
|
||||
*timeOut = atoi (argv[4]); /* The number of seconds the publisher should run for (0 = infinite) */
|
||||
}
|
||||
if (argc > 5)
|
||||
{
|
||||
*partitionName = argv[5]; /* The name of the partition */
|
||||
}
|
||||
|
||||
printf ("payloadSize: %u bytes burstInterval: %u ms burstSize: %u timeOut: %u seconds partitionName: %s\n",
|
||||
*payloadSize, *burstInterval, *burstSize, *timeOut, *partitionName);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
static dds_entity_t prepare_dds(dds_entity_t *writer, const char *partitionName)
|
||||
{
|
||||
dds_entity_t participant;
|
||||
dds_entity_t topic;
|
||||
dds_entity_t publisher;
|
||||
const char *pubParts[1];
|
||||
dds_qos_t *pubQos;
|
||||
dds_qos_t *dwQos;
|
||||
|
||||
/* A domain participant is created for the default domain. */
|
||||
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
|
||||
DDS_ERR_CHECK (participant, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A topic is created for our sample type on the domain participant. */
|
||||
topic = dds_create_topic (participant, &ThroughputModule_DataType_desc, "Throughput", NULL, NULL);
|
||||
DDS_ERR_CHECK (topic, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A publisher is created on the domain participant. */
|
||||
pubQos = dds_qos_create ();
|
||||
pubParts[0] = partitionName;
|
||||
dds_qset_partition (pubQos, 1, pubParts);
|
||||
publisher = dds_create_publisher (participant, pubQos, NULL);
|
||||
DDS_ERR_CHECK (publisher, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (pubQos);
|
||||
|
||||
/* A DataWriter is created on the publisher. */
|
||||
dwQos = dds_qos_create ();
|
||||
dds_qset_reliability (dwQos, DDS_RELIABILITY_RELIABLE, DDS_SECS (10));
|
||||
dds_qset_history (dwQos, DDS_HISTORY_KEEP_ALL, 0);
|
||||
dds_qset_resource_limits (dwQos, MAX_SAMPLES, DDS_LENGTH_UNLIMITED, DDS_LENGTH_UNLIMITED);
|
||||
*writer = dds_create_writer (publisher, topic, dwQos, NULL);
|
||||
DDS_ERR_CHECK (*writer, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (dwQos);
|
||||
|
||||
/* Enable write batching */
|
||||
dds_write_set_batch (true);
|
||||
|
||||
return participant;
|
||||
}
|
||||
|
||||
static dds_return_t wait_for_reader(dds_entity_t writer, dds_entity_t participant)
|
||||
{
|
||||
printf ("\n=== [Publisher] Waiting for a reader ...\n");
|
||||
|
||||
dds_return_t ret;
|
||||
dds_entity_t waitset;
|
||||
|
||||
ret = dds_set_enabled_status(writer, DDS_PUBLICATION_MATCHED_STATUS);
|
||||
DDS_ERR_CHECK (ret, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
waitset = dds_create_waitset(participant);
|
||||
DDS_ERR_CHECK (waitset, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
ret = dds_waitset_attach(waitset, writer, (dds_attach_t)NULL);
|
||||
DDS_ERR_CHECK (waitset, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
ret = dds_waitset_wait(waitset, NULL, 0, DDS_SECS(30));
|
||||
DDS_ERR_CHECK (ret, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
static void start_writing(
|
||||
dds_entity_t writer,
|
||||
ThroughputModule_DataType *sample,
|
||||
unsigned int burstInterval,
|
||||
unsigned int burstSize,
|
||||
unsigned int timeOut)
|
||||
{
|
||||
bool timedOut = false;
|
||||
dds_time_t pubStart = dds_time ();
|
||||
dds_time_t now;
|
||||
dds_time_t deltaTv;
|
||||
dds_return_t status;
|
||||
|
||||
if (!done)
|
||||
{
|
||||
dds_time_t burstStart = pubStart;
|
||||
unsigned int burstCount = 0;
|
||||
|
||||
printf ("=== [Publisher] Writing samples...\n");
|
||||
|
||||
while (!done && !timedOut)
|
||||
{
|
||||
/* Write data until burst size has been reached */
|
||||
|
||||
if (burstCount < burstSize)
|
||||
{
|
||||
status = dds_write (writer, sample);
|
||||
if (dds_err_nr(status) == DDS_RETCODE_TIMEOUT)
|
||||
{
|
||||
timedOut = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
sample->count++;
|
||||
burstCount++;
|
||||
}
|
||||
}
|
||||
else if (burstInterval)
|
||||
{
|
||||
/* Sleep until burst interval has passed */
|
||||
|
||||
dds_time_t time = dds_time ();
|
||||
deltaTv = time - burstStart;
|
||||
if (deltaTv < DDS_MSECS (burstInterval))
|
||||
{
|
||||
dds_write_flush (writer);
|
||||
dds_sleepfor (DDS_MSECS (burstInterval) - deltaTv);
|
||||
}
|
||||
burstStart = dds_time ();
|
||||
burstCount = 0;
|
||||
}
|
||||
else
|
||||
{
|
||||
burstCount = 0;
|
||||
}
|
||||
|
||||
if (timeOut)
|
||||
{
|
||||
now = dds_time ();
|
||||
deltaTv = now - pubStart;
|
||||
if ((deltaTv) > DDS_SECS (timeOut))
|
||||
{
|
||||
timedOut = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
dds_write_flush (writer);
|
||||
|
||||
if (done)
|
||||
{
|
||||
printf ("=== [Publisher] Terminated, %llu samples written.\n", (unsigned long long) sample->count);
|
||||
}
|
||||
else
|
||||
{
|
||||
printf ("=== [Publisher] Timed out, %llu samples written.\n", (unsigned long long) sample->count);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void finalize_dds(dds_entity_t participant, dds_entity_t writer, ThroughputModule_DataType sample)
|
||||
{
|
||||
dds_return_t status = dds_dispose (writer, &sample);
|
||||
if (dds_err_nr (status) != DDS_RETCODE_TIMEOUT)
|
||||
{
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
}
|
||||
|
||||
dds_free (sample.payload._buffer);
|
||||
status = dds_delete (participant);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
}
|
||||
#include "ddsc/dds.h"
|
||||
#include "Throughput.h"
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <signal.h>
|
||||
|
||||
/*
|
||||
* The Throughput example measures data throughput in bytes per second. The publisher
|
||||
* allows you to specify a payload size in bytes as well as allowing you to specify
|
||||
* whether to send data in bursts. The publisher will continue to send data forever
|
||||
* unless a time out is specified. The subscriber will receive data and output the
|
||||
* total amount received and the data rate in bytes per second. It will also indicate
|
||||
* if any samples were received out of order. A maximum number of cycles can be
|
||||
* specified and once this has been reached the subscriber will terminate and output
|
||||
* totals and averages.
|
||||
*/
|
||||
|
||||
#define MAX_SAMPLES 100
|
||||
|
||||
static bool done = false;
|
||||
|
||||
/* Forward declarations */
|
||||
static dds_return_t wait_for_reader(dds_entity_t writer, dds_entity_t participant);
|
||||
static void start_writing(dds_entity_t writer, ThroughputModule_DataType *sample,
|
||||
int burstInterval, uint32_t burstSize, int timeOut);
|
||||
static int parse_args(int argc, char **argv, uint32_t *payloadSize, int *burstInterval,
|
||||
uint32_t *burstSize, int *timeOut, char **partitionName);
|
||||
static dds_entity_t prepare_dds(dds_entity_t *writer, const char *partitionName);
|
||||
static void finalize_dds(dds_entity_t participant, dds_entity_t writer, ThroughputModule_DataType sample);
|
||||
|
||||
/* Functions to handle Ctrl-C presses. */
|
||||
#ifdef _WIN32
|
||||
#include <Windows.h>
|
||||
static int CtrlHandler (DWORD fdwCtrlType)
|
||||
{
|
||||
done = true;
|
||||
return true; /* Don't let other handlers handle this key */
|
||||
}
|
||||
#else
|
||||
struct sigaction oldAction;
|
||||
static void CtrlHandler (int sig)
|
||||
{
|
||||
(void)sig;
|
||||
done = true;
|
||||
}
|
||||
#endif
|
||||
|
||||
int main (int argc, char **argv)
|
||||
{
|
||||
uint32_t payloadSize = 8192;
|
||||
int burstInterval = 0;
|
||||
uint32_t burstSize = 1;
|
||||
int timeOut = 0;
|
||||
char * partitionName = "Throughput example";
|
||||
dds_entity_t participant;
|
||||
dds_entity_t writer;
|
||||
ThroughputModule_DataType sample;
|
||||
|
||||
/* Register handler for Ctrl-C */
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler ((PHANDLER_ROUTINE) CtrlHandler, true);
|
||||
#else
|
||||
struct sigaction sat;
|
||||
sat.sa_handler = CtrlHandler;
|
||||
sigemptyset (&sat.sa_mask);
|
||||
sat.sa_flags = 0;
|
||||
sigaction (SIGINT, &sat, &oldAction);
|
||||
#endif
|
||||
|
||||
if (parse_args(argc, argv, &payloadSize, &burstInterval, &burstSize, &timeOut, &partitionName) == EXIT_FAILURE) {
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
participant = prepare_dds(&writer, partitionName);
|
||||
|
||||
/* Wait until have a reader */
|
||||
if (wait_for_reader(writer, participant) == 0) {
|
||||
printf ("=== [Publisher] Did not discover a reader.\n");
|
||||
DDS_ERR_CHECK (dds_delete (participant), DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
/* Fill the sample payload with data */
|
||||
sample.count = 0;
|
||||
sample.payload._buffer = dds_alloc (payloadSize);
|
||||
sample.payload._length = payloadSize;
|
||||
sample.payload._release = true;
|
||||
for (uint32_t i = 0; i < payloadSize; i++) {
|
||||
sample.payload._buffer[i] = 'a';
|
||||
}
|
||||
|
||||
/* Register the sample instance and write samples repeatedly or until time out */
|
||||
start_writing(writer, &sample, burstInterval, burstSize, timeOut);
|
||||
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler (0, false);
|
||||
#else
|
||||
sigaction (SIGINT, &oldAction, 0);
|
||||
#endif
|
||||
|
||||
/* Cleanup */
|
||||
finalize_dds(participant, writer, sample);
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
static int parse_args(
|
||||
int argc,
|
||||
char **argv,
|
||||
uint32_t *payloadSize,
|
||||
int *burstInterval,
|
||||
uint32_t *burstSize,
|
||||
int *timeOut,
|
||||
char **partitionName)
|
||||
{
|
||||
int result = EXIT_SUCCESS;
|
||||
/*
|
||||
* Get the program parameters
|
||||
* Parameters: publisher [payloadSize] [burstInterval] [burstSize] [timeOut] [partitionName]
|
||||
*/
|
||||
if (argc == 2 && (strcmp (argv[1], "-h") == 0 || strcmp (argv[1], "--help") == 0))
|
||||
{
|
||||
printf ("Usage (parameters must be supplied in order):\n");
|
||||
printf ("./publisher [payloadSize (bytes)] [burstInterval (ms)] [burstSize (samples)] [timeOut (seconds)] [partitionName]\n");
|
||||
printf ("Defaults:\n");
|
||||
printf ("./publisher 8192 0 1 0 \"Throughput example\"\n");
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
if (argc > 1)
|
||||
{
|
||||
*payloadSize = (uint32_t) atoi (argv[1]); /* The size of the payload in bytes */
|
||||
}
|
||||
if (argc > 2)
|
||||
{
|
||||
*burstInterval = atoi (argv[2]); /* The time interval between each burst in ms */
|
||||
}
|
||||
if (argc > 3)
|
||||
{
|
||||
*burstSize = (uint32_t) atoi (argv[3]); /* The number of samples to send each burst */
|
||||
}
|
||||
if (argc > 4)
|
||||
{
|
||||
*timeOut = atoi (argv[4]); /* The number of seconds the publisher should run for (0 = infinite) */
|
||||
}
|
||||
if (argc > 5)
|
||||
{
|
||||
*partitionName = argv[5]; /* The name of the partition */
|
||||
}
|
||||
|
||||
printf ("payloadSize: %u bytes burstInterval: %u ms burstSize: %u timeOut: %u seconds partitionName: %s\n",
|
||||
*payloadSize, *burstInterval, *burstSize, *timeOut, *partitionName);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
static dds_entity_t prepare_dds(dds_entity_t *writer, const char *partitionName)
|
||||
{
|
||||
dds_entity_t participant;
|
||||
dds_entity_t topic;
|
||||
dds_entity_t publisher;
|
||||
const char *pubParts[1];
|
||||
dds_qos_t *pubQos;
|
||||
dds_qos_t *dwQos;
|
||||
|
||||
/* A domain participant is created for the default domain. */
|
||||
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
|
||||
DDS_ERR_CHECK (participant, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A topic is created for our sample type on the domain participant. */
|
||||
topic = dds_create_topic (participant, &ThroughputModule_DataType_desc, "Throughput", NULL, NULL);
|
||||
DDS_ERR_CHECK (topic, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A publisher is created on the domain participant. */
|
||||
pubQos = dds_qos_create ();
|
||||
pubParts[0] = partitionName;
|
||||
dds_qset_partition (pubQos, 1, pubParts);
|
||||
publisher = dds_create_publisher (participant, pubQos, NULL);
|
||||
DDS_ERR_CHECK (publisher, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (pubQos);
|
||||
|
||||
/* A DataWriter is created on the publisher. */
|
||||
dwQos = dds_qos_create ();
|
||||
dds_qset_reliability (dwQos, DDS_RELIABILITY_RELIABLE, DDS_SECS (10));
|
||||
dds_qset_history (dwQos, DDS_HISTORY_KEEP_ALL, 0);
|
||||
dds_qset_resource_limits (dwQos, MAX_SAMPLES, DDS_LENGTH_UNLIMITED, DDS_LENGTH_UNLIMITED);
|
||||
*writer = dds_create_writer (publisher, topic, dwQos, NULL);
|
||||
DDS_ERR_CHECK (*writer, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (dwQos);
|
||||
|
||||
/* Enable write batching */
|
||||
dds_write_set_batch (true);
|
||||
|
||||
return participant;
|
||||
}
|
||||
|
||||
static dds_return_t wait_for_reader(dds_entity_t writer, dds_entity_t participant)
|
||||
{
|
||||
printf ("\n=== [Publisher] Waiting for a reader ...\n");
|
||||
|
||||
dds_return_t ret;
|
||||
dds_entity_t waitset;
|
||||
|
||||
ret = dds_set_enabled_status(writer, DDS_PUBLICATION_MATCHED_STATUS);
|
||||
DDS_ERR_CHECK (ret, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
waitset = dds_create_waitset(participant);
|
||||
DDS_ERR_CHECK (waitset, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
ret = dds_waitset_attach(waitset, writer, (dds_attach_t)NULL);
|
||||
DDS_ERR_CHECK (waitset, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
ret = dds_waitset_wait(waitset, NULL, 0, DDS_SECS(30));
|
||||
DDS_ERR_CHECK (ret, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
static void start_writing(
|
||||
dds_entity_t writer,
|
||||
ThroughputModule_DataType *sample,
|
||||
int burstInterval,
|
||||
uint32_t burstSize,
|
||||
int timeOut)
|
||||
{
|
||||
bool timedOut = false;
|
||||
dds_time_t pubStart = dds_time ();
|
||||
dds_time_t now;
|
||||
dds_time_t deltaTv;
|
||||
dds_return_t status;
|
||||
|
||||
if (!done)
|
||||
{
|
||||
dds_time_t burstStart = pubStart;
|
||||
unsigned int burstCount = 0;
|
||||
|
||||
printf ("=== [Publisher] Writing samples...\n");
|
||||
|
||||
while (!done && !timedOut)
|
||||
{
|
||||
/* Write data until burst size has been reached */
|
||||
|
||||
if (burstCount < burstSize)
|
||||
{
|
||||
status = dds_write (writer, sample);
|
||||
if (dds_err_nr(status) == DDS_RETCODE_TIMEOUT)
|
||||
{
|
||||
timedOut = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
sample->count++;
|
||||
burstCount++;
|
||||
}
|
||||
}
|
||||
else if (burstInterval)
|
||||
{
|
||||
/* Sleep until burst interval has passed */
|
||||
|
||||
dds_time_t time = dds_time ();
|
||||
deltaTv = time - burstStart;
|
||||
if (deltaTv < DDS_MSECS (burstInterval))
|
||||
{
|
||||
dds_write_flush (writer);
|
||||
dds_sleepfor (DDS_MSECS (burstInterval) - deltaTv);
|
||||
}
|
||||
burstStart = dds_time ();
|
||||
burstCount = 0;
|
||||
}
|
||||
else
|
||||
{
|
||||
burstCount = 0;
|
||||
}
|
||||
|
||||
if (timeOut)
|
||||
{
|
||||
now = dds_time ();
|
||||
deltaTv = now - pubStart;
|
||||
if ((deltaTv) > DDS_SECS (timeOut))
|
||||
{
|
||||
timedOut = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
dds_write_flush (writer);
|
||||
|
||||
if (done)
|
||||
{
|
||||
printf ("=== [Publisher] Terminated, %llu samples written.\n", (unsigned long long) sample->count);
|
||||
}
|
||||
else
|
||||
{
|
||||
printf ("=== [Publisher] Timed out, %llu samples written.\n", (unsigned long long) sample->count);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void finalize_dds(dds_entity_t participant, dds_entity_t writer, ThroughputModule_DataType sample)
|
||||
{
|
||||
dds_return_t status = dds_dispose (writer, &sample);
|
||||
if (dds_err_nr (status) != DDS_RETCODE_TIMEOUT)
|
||||
{
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
}
|
||||
|
||||
dds_free (sample.payload._buffer);
|
||||
status = dds_delete (participant);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,426 +1,428 @@
|
|||
#include "ddsc/dds.h"
|
||||
#include "Throughput.h"
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <signal.h>
|
||||
#include <assert.h>
|
||||
/*
|
||||
* The Throughput example measures data throughput in bytes per second. The publisher
|
||||
* allows you to specify a payload size in bytes as well as allowing you to specify
|
||||
* whether to send data in bursts. The publisher will continue to send data forever
|
||||
* unless a time out is specified. The subscriber will receive data and output the
|
||||
* total amount received and the data rate in bytes per second. It will also indicate
|
||||
* if any samples were received out of order. A maximum number of cycles can be
|
||||
* specified and once this has been reached the subscriber will terminate and output
|
||||
* totals and averages.
|
||||
*/
|
||||
|
||||
#define BYTES_PER_SEC_TO_MEGABITS_PER_SEC 125000
|
||||
#define MAX_SAMPLES 100
|
||||
|
||||
typedef struct HandleEntry
|
||||
{
|
||||
dds_instance_handle_t handle;
|
||||
unsigned long long count;
|
||||
struct HandleEntry * next;
|
||||
} HandleEntry;
|
||||
|
||||
typedef struct HandleMap
|
||||
{
|
||||
HandleEntry *entries;
|
||||
} HandleMap;
|
||||
|
||||
static unsigned long pollingDelay = 0;
|
||||
|
||||
static HandleMap * imap;
|
||||
static unsigned long long outOfOrder = 0;
|
||||
static unsigned long long total_bytes = 0;
|
||||
static unsigned long long total_samples = 0;
|
||||
|
||||
static dds_time_t startTime = 0;
|
||||
static dds_time_t time_now = 0;
|
||||
static dds_time_t prev_time = 0;
|
||||
|
||||
static unsigned long payloadSize = 0;
|
||||
|
||||
static ThroughputModule_DataType data [MAX_SAMPLES];
|
||||
static void * samples[MAX_SAMPLES];
|
||||
|
||||
static dds_entity_t waitSet;
|
||||
static dds_entity_t pollingWaitset;
|
||||
|
||||
static bool done = false;
|
||||
|
||||
/* Forward declarations */
|
||||
static HandleMap * HandleMap__alloc (void);
|
||||
static void HandleMap__free (HandleMap *map);
|
||||
static HandleEntry * store_handle (HandleMap *map, dds_instance_handle_t key);
|
||||
static HandleEntry * retrieve_handle (HandleMap *map, dds_instance_handle_t key);
|
||||
|
||||
static void data_available_handler (dds_entity_t reader, void *arg);
|
||||
static int parse_args(int argc, char **argv, unsigned long long *maxCycles, char **partitionName);
|
||||
static void process_samples(unsigned long long maxCycles);
|
||||
static dds_entity_t prepare_dds(dds_entity_t *reader, const char *partitionName);
|
||||
static void finalize_dds(dds_entity_t participant);
|
||||
|
||||
/* Functions to handle Ctrl-C presses. */
|
||||
#ifdef _WIN32
|
||||
#include <Windows.h>
|
||||
static int CtrlHandler (DWORD fdwCtrlType)
|
||||
{
|
||||
dds_waitset_set_trigger (waitSet, true);
|
||||
done = true;
|
||||
return true; /* Don't let other handlers handle this key */
|
||||
}
|
||||
#else
|
||||
struct sigaction oldAction;
|
||||
static void CtrlHandler (int fdwCtrlType)
|
||||
{
|
||||
dds_waitset_set_trigger (waitSet, true);
|
||||
done = true;
|
||||
}
|
||||
#endif
|
||||
|
||||
int main (int argc, char **argv)
|
||||
{
|
||||
unsigned long long maxCycles = 0;
|
||||
char *partitionName = "Throughput example";
|
||||
|
||||
dds_entity_t participant;
|
||||
dds_entity_t reader;
|
||||
|
||||
time_now = dds_time ();
|
||||
prev_time = time_now;
|
||||
|
||||
/* Register handler for Ctrl-C */
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler((PHANDLER_ROUTINE) CtrlHandler, true);
|
||||
#else
|
||||
struct sigaction sat;
|
||||
sat.sa_handler = CtrlHandler;
|
||||
sigemptyset(&sat.sa_mask);
|
||||
sat.sa_flags = 0;
|
||||
sigaction (SIGINT, &sat, &oldAction);
|
||||
#endif
|
||||
|
||||
if (parse_args(argc, argv, &maxCycles, &partitionName) == EXIT_FAILURE)
|
||||
{
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
printf ("Cycles: %llu | PollingDelay: %lu | Partition: %s\n",
|
||||
maxCycles, pollingDelay, partitionName);
|
||||
|
||||
participant = prepare_dds(&reader, partitionName);
|
||||
|
||||
printf ("=== [Subscriber] Waiting for samples...\n");
|
||||
|
||||
/* Process samples until Ctrl-C is pressed or until maxCycles */
|
||||
/* has been reached (0 = infinite) */
|
||||
process_samples(maxCycles);
|
||||
|
||||
/* Finished, disable callbacks */
|
||||
dds_set_enabled_status (reader, 0);
|
||||
HandleMap__free (imap);
|
||||
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler (0, FALSE);
|
||||
#else
|
||||
sigaction (SIGINT, &oldAction, 0);
|
||||
#endif
|
||||
|
||||
/* Clean up */
|
||||
finalize_dds(participant);
|
||||
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* This struct contains all of the entities used in the publisher and subscriber.
|
||||
*/
|
||||
static HandleMap * HandleMap__alloc (void)
|
||||
{
|
||||
HandleMap * map = malloc (sizeof (*map));
|
||||
assert(map);
|
||||
memset (map, 0, sizeof (*map));
|
||||
return map;
|
||||
}
|
||||
|
||||
static void HandleMap__free (HandleMap *map)
|
||||
{
|
||||
HandleEntry * entry;
|
||||
|
||||
while (map->entries)
|
||||
{
|
||||
entry = map->entries;
|
||||
map->entries = entry->next;
|
||||
free (entry);
|
||||
}
|
||||
free (map);
|
||||
}
|
||||
|
||||
static HandleEntry * store_handle (HandleMap *map, dds_instance_handle_t key)
|
||||
{
|
||||
HandleEntry * entry = malloc (sizeof (*entry));
|
||||
assert(entry);
|
||||
memset (entry, 0, sizeof (*entry));
|
||||
|
||||
entry->handle = key;
|
||||
entry->next = map->entries;
|
||||
map->entries = entry;
|
||||
|
||||
return entry;
|
||||
}
|
||||
|
||||
static HandleEntry * retrieve_handle (HandleMap *map, dds_instance_handle_t key)
|
||||
{
|
||||
HandleEntry * entry = map->entries;
|
||||
|
||||
while (entry)
|
||||
{
|
||||
if (entry->handle == key)
|
||||
{
|
||||
break;
|
||||
}
|
||||
entry = entry->next;
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
|
||||
static void data_available_handler (dds_entity_t reader, void *arg)
|
||||
{
|
||||
int samples_received;
|
||||
dds_sample_info_t info [MAX_SAMPLES];
|
||||
dds_instance_handle_t ph = 0;
|
||||
HandleEntry * current = NULL;
|
||||
|
||||
if (startTime == 0)
|
||||
{
|
||||
startTime = dds_time ();
|
||||
}
|
||||
|
||||
/* Take samples and iterate through them */
|
||||
|
||||
samples_received = dds_take (reader, samples, info, MAX_SAMPLES, MAX_SAMPLES);
|
||||
DDS_ERR_CHECK (samples_received, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
for (int i = 0; !done && i < samples_received; i++)
|
||||
{
|
||||
if (info[i].valid_data)
|
||||
{
|
||||
ph = info[i].publication_handle;
|
||||
current = retrieve_handle (imap, ph);
|
||||
ThroughputModule_DataType * this_sample = &data[i];
|
||||
|
||||
if (current == NULL)
|
||||
{
|
||||
current = store_handle (imap, ph);
|
||||
current->count = this_sample->count;
|
||||
}
|
||||
|
||||
if (this_sample->count != current->count)
|
||||
{
|
||||
outOfOrder++;
|
||||
}
|
||||
current->count = this_sample->count + 1;
|
||||
|
||||
/* Add the sample payload size to the total received */
|
||||
|
||||
payloadSize = this_sample->payload._length;
|
||||
total_bytes += payloadSize + 8;
|
||||
total_samples++;
|
||||
}
|
||||
}
|
||||
time_now = dds_time ();
|
||||
if ((pollingDelay == 0) && (time_now > (prev_time + DDS_SECS (1))))
|
||||
{
|
||||
dds_waitset_set_trigger (pollingWaitset, true);
|
||||
}
|
||||
}
|
||||
|
||||
static int parse_args(int argc, char **argv, unsigned long long *maxCycles, char **partitionName)
|
||||
{
|
||||
/*
|
||||
* Get the program parameters
|
||||
* Parameters: subscriber [maxCycles] [pollingDelay] [partitionName]
|
||||
*/
|
||||
if (argc == 2 && (strcmp (argv[1], "-h") == 0 || strcmp (argv[1], "--help") == 0))
|
||||
{
|
||||
printf ("Usage (parameters must be supplied in order):\n");
|
||||
printf ("./subscriber [maxCycles (0 = infinite)] [pollingDelay (ms, 0 = event based)] [partitionName]\n");
|
||||
printf ("Defaults:\n");
|
||||
printf ("./subscriber 0 0 \"Throughput example\"\n");
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
if (argc > 1)
|
||||
{
|
||||
*maxCycles = atoi (argv[1]); /* The number of times to output statistics before terminating */
|
||||
}
|
||||
if (argc > 2)
|
||||
{
|
||||
pollingDelay = atoi (argv[2]); /* The number of ms to wait between reads (0 = event based) */
|
||||
}
|
||||
if (argc > 3)
|
||||
{
|
||||
*partitionName = argv[3]; /* The name of the partition */
|
||||
}
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
static void process_samples(unsigned long long maxCycles)
|
||||
{
|
||||
dds_return_t status;
|
||||
unsigned long long prev_bytes = 0;
|
||||
unsigned long long prev_samples = 0;
|
||||
dds_attach_t wsresults[1];
|
||||
size_t wsresultsize = 1U;
|
||||
dds_time_t deltaTv;
|
||||
bool first_batch = true;
|
||||
unsigned long cycles = 0;
|
||||
double deltaTime = 0;
|
||||
|
||||
while (!done && (maxCycles == 0 || cycles < maxCycles))
|
||||
{
|
||||
if (pollingDelay)
|
||||
{
|
||||
dds_sleepfor (DDS_MSECS (pollingDelay));
|
||||
}
|
||||
else
|
||||
{
|
||||
status = dds_waitset_wait (waitSet, wsresults, wsresultsize, DDS_INFINITY);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
if ((status > 0 ) && (dds_triggered (pollingWaitset)))
|
||||
{
|
||||
dds_waitset_set_trigger (pollingWaitset, false);
|
||||
}
|
||||
}
|
||||
|
||||
if (!first_batch)
|
||||
{
|
||||
deltaTv = time_now - prev_time;
|
||||
deltaTime = (double) deltaTv / DDS_NSECS_IN_SEC;
|
||||
prev_time = time_now;
|
||||
|
||||
printf
|
||||
(
|
||||
"=== [Subscriber] Payload size: %lu | Total received: %llu samples, %llu bytes | Out of order: %llu samples "
|
||||
"Transfer rate: %.2lf samples/s, %.2lf Mbit/s\n",
|
||||
payloadSize, total_samples, total_bytes, outOfOrder,
|
||||
(deltaTime) ? ((total_samples - prev_samples) / deltaTime) : 0,
|
||||
(deltaTime) ? (((total_bytes - prev_bytes) / BYTES_PER_SEC_TO_MEGABITS_PER_SEC) / deltaTime) : 0
|
||||
);
|
||||
cycles++;
|
||||
}
|
||||
else
|
||||
{
|
||||
prev_time = time_now;
|
||||
first_batch = false;
|
||||
}
|
||||
|
||||
/* Update the previous values for next iteration */
|
||||
|
||||
prev_bytes = total_bytes;
|
||||
prev_samples = total_samples;
|
||||
}
|
||||
|
||||
/* Output totals and averages */
|
||||
deltaTv = time_now - startTime;
|
||||
deltaTime = (double) (deltaTv / DDS_NSECS_IN_SEC);
|
||||
printf ("\nTotal received: %llu samples, %llu bytes\n", total_samples, total_bytes);
|
||||
printf ("Out of order: %llu samples\n", outOfOrder);
|
||||
printf ("Average transfer rate: %.2lf samples/s, ", total_samples / deltaTime);
|
||||
printf ("%.2lf Mbit/s\n", (total_bytes / BYTES_PER_SEC_TO_MEGABITS_PER_SEC) / deltaTime);
|
||||
}
|
||||
|
||||
static dds_entity_t prepare_dds(dds_entity_t *reader, const char *partitionName)
|
||||
{
|
||||
dds_return_t status;
|
||||
dds_entity_t topic;
|
||||
dds_entity_t subscriber;
|
||||
dds_listener_t *rd_listener;
|
||||
dds_entity_t participant;
|
||||
|
||||
uint32_t maxSamples = 400;
|
||||
const char *subParts[1];
|
||||
dds_qos_t *subQos = dds_qos_create ();
|
||||
dds_qos_t *drQos = dds_qos_create ();
|
||||
|
||||
/* A Participant is created for the default domain. */
|
||||
|
||||
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
|
||||
DDS_ERR_CHECK (participant, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A Topic is created for our sample type on the domain participant. */
|
||||
|
||||
topic = dds_create_topic (participant, &ThroughputModule_DataType_desc, "Throughput", NULL, NULL);
|
||||
DDS_ERR_CHECK (topic, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A Subscriber is created on the domain participant. */
|
||||
|
||||
subParts[0] = partitionName;
|
||||
dds_qset_partition (subQos, 1, subParts);
|
||||
subscriber = dds_create_subscriber (participant, subQos, NULL);
|
||||
DDS_ERR_CHECK (subscriber, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (subQos);
|
||||
|
||||
/* A Reader is created on the Subscriber & Topic with a modified Qos. */
|
||||
|
||||
dds_qset_reliability (drQos, DDS_RELIABILITY_RELIABLE, DDS_SECS (10));
|
||||
dds_qset_history (drQos, DDS_HISTORY_KEEP_ALL, 0);
|
||||
dds_qset_resource_limits (drQos, maxSamples, DDS_LENGTH_UNLIMITED, DDS_LENGTH_UNLIMITED);
|
||||
|
||||
rd_listener = dds_listener_create(NULL);
|
||||
dds_lset_data_available(rd_listener, data_available_handler);
|
||||
|
||||
/* A Read Condition is created which is triggered when data is available to read */
|
||||
waitSet = dds_create_waitset (participant);
|
||||
DDS_ERR_CHECK (waitSet, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
pollingWaitset = dds_create_waitset (participant);
|
||||
DDS_ERR_CHECK (pollingWaitset, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
status = dds_waitset_attach (waitSet, pollingWaitset, pollingWaitset);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
status = dds_waitset_attach (waitSet, waitSet, waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
imap = HandleMap__alloc ();
|
||||
|
||||
memset (data, 0, sizeof (data));
|
||||
for (unsigned int i = 0; i < MAX_SAMPLES; i++)
|
||||
{
|
||||
samples[i] = &data[i];
|
||||
}
|
||||
|
||||
*reader = dds_create_reader (subscriber, topic, drQos, rd_listener);
|
||||
DDS_ERR_CHECK (*reader, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (drQos);
|
||||
dds_listener_delete(rd_listener);
|
||||
|
||||
return participant;
|
||||
}
|
||||
|
||||
static void finalize_dds(dds_entity_t participant)
|
||||
{
|
||||
dds_return_t status;
|
||||
|
||||
for (unsigned int i = 0; i < MAX_SAMPLES; i++)
|
||||
{
|
||||
ThroughputModule_DataType_free (&data[i], DDS_FREE_CONTENTS);
|
||||
}
|
||||
|
||||
status = dds_waitset_detach (waitSet, waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_waitset_detach (waitSet, pollingWaitset);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_delete (pollingWaitset);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_delete (waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_delete (participant);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
}
|
||||
#include "ddsc/dds.h"
|
||||
#include "Throughput.h"
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <signal.h>
|
||||
#include <assert.h>
|
||||
/*
|
||||
* The Throughput example measures data throughput in bytes per second. The publisher
|
||||
* allows you to specify a payload size in bytes as well as allowing you to specify
|
||||
* whether to send data in bursts. The publisher will continue to send data forever
|
||||
* unless a time out is specified. The subscriber will receive data and output the
|
||||
* total amount received and the data rate in bytes per second. It will also indicate
|
||||
* if any samples were received out of order. A maximum number of cycles can be
|
||||
* specified and once this has been reached the subscriber will terminate and output
|
||||
* totals and averages.
|
||||
*/
|
||||
|
||||
#define BYTES_PER_SEC_TO_MEGABITS_PER_SEC 125000
|
||||
#define MAX_SAMPLES 100
|
||||
|
||||
typedef struct HandleEntry
|
||||
{
|
||||
dds_instance_handle_t handle;
|
||||
unsigned long long count;
|
||||
struct HandleEntry * next;
|
||||
} HandleEntry;
|
||||
|
||||
typedef struct HandleMap
|
||||
{
|
||||
HandleEntry *entries;
|
||||
} HandleMap;
|
||||
|
||||
static long pollingDelay = 0;
|
||||
|
||||
static HandleMap * imap;
|
||||
static unsigned long long outOfOrder = 0;
|
||||
static unsigned long long total_bytes = 0;
|
||||
static unsigned long long total_samples = 0;
|
||||
|
||||
static dds_time_t startTime = 0;
|
||||
static dds_time_t time_now = 0;
|
||||
static dds_time_t prev_time = 0;
|
||||
|
||||
static unsigned long payloadSize = 0;
|
||||
|
||||
static ThroughputModule_DataType data [MAX_SAMPLES];
|
||||
static void * samples[MAX_SAMPLES];
|
||||
|
||||
static dds_entity_t waitSet;
|
||||
static dds_entity_t pollingWaitset;
|
||||
|
||||
static bool done = false;
|
||||
|
||||
/* Forward declarations */
|
||||
static HandleMap * HandleMap__alloc (void);
|
||||
static void HandleMap__free (HandleMap *map);
|
||||
static HandleEntry * store_handle (HandleMap *map, dds_instance_handle_t key);
|
||||
static HandleEntry * retrieve_handle (HandleMap *map, dds_instance_handle_t key);
|
||||
|
||||
static void data_available_handler (dds_entity_t reader, void *arg);
|
||||
static int parse_args(int argc, char **argv, unsigned long long *maxCycles, char **partitionName);
|
||||
static void process_samples(unsigned long long maxCycles);
|
||||
static dds_entity_t prepare_dds(dds_entity_t *reader, const char *partitionName);
|
||||
static void finalize_dds(dds_entity_t participant);
|
||||
|
||||
/* Functions to handle Ctrl-C presses. */
|
||||
#ifdef _WIN32
|
||||
#include <Windows.h>
|
||||
static int CtrlHandler (DWORD fdwCtrlType)
|
||||
{
|
||||
dds_waitset_set_trigger (waitSet, true);
|
||||
done = true;
|
||||
return true; /* Don't let other handlers handle this key */
|
||||
}
|
||||
#else
|
||||
struct sigaction oldAction;
|
||||
static void CtrlHandler (int sig)
|
||||
{
|
||||
(void)sig;
|
||||
dds_waitset_set_trigger (waitSet, true);
|
||||
done = true;
|
||||
}
|
||||
#endif
|
||||
|
||||
int main (int argc, char **argv)
|
||||
{
|
||||
unsigned long long maxCycles = 0;
|
||||
char *partitionName = "Throughput example";
|
||||
|
||||
dds_entity_t participant;
|
||||
dds_entity_t reader;
|
||||
|
||||
time_now = dds_time ();
|
||||
prev_time = time_now;
|
||||
|
||||
/* Register handler for Ctrl-C */
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler((PHANDLER_ROUTINE) CtrlHandler, true);
|
||||
#else
|
||||
struct sigaction sat;
|
||||
sat.sa_handler = CtrlHandler;
|
||||
sigemptyset(&sat.sa_mask);
|
||||
sat.sa_flags = 0;
|
||||
sigaction (SIGINT, &sat, &oldAction);
|
||||
#endif
|
||||
|
||||
if (parse_args(argc, argv, &maxCycles, &partitionName) == EXIT_FAILURE)
|
||||
{
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
printf ("Cycles: %llu | PollingDelay: %lu | Partition: %s\n",
|
||||
maxCycles, pollingDelay, partitionName);
|
||||
|
||||
participant = prepare_dds(&reader, partitionName);
|
||||
|
||||
printf ("=== [Subscriber] Waiting for samples...\n");
|
||||
|
||||
/* Process samples until Ctrl-C is pressed or until maxCycles */
|
||||
/* has been reached (0 = infinite) */
|
||||
process_samples(maxCycles);
|
||||
|
||||
/* Finished, disable callbacks */
|
||||
dds_set_enabled_status (reader, 0);
|
||||
HandleMap__free (imap);
|
||||
|
||||
#ifdef _WIN32
|
||||
SetConsoleCtrlHandler (0, FALSE);
|
||||
#else
|
||||
sigaction (SIGINT, &oldAction, 0);
|
||||
#endif
|
||||
|
||||
/* Clean up */
|
||||
finalize_dds(participant);
|
||||
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* This struct contains all of the entities used in the publisher and subscriber.
|
||||
*/
|
||||
static HandleMap * HandleMap__alloc (void)
|
||||
{
|
||||
HandleMap * map = malloc (sizeof (*map));
|
||||
assert(map);
|
||||
memset (map, 0, sizeof (*map));
|
||||
return map;
|
||||
}
|
||||
|
||||
static void HandleMap__free (HandleMap *map)
|
||||
{
|
||||
HandleEntry * entry;
|
||||
|
||||
while (map->entries)
|
||||
{
|
||||
entry = map->entries;
|
||||
map->entries = entry->next;
|
||||
free (entry);
|
||||
}
|
||||
free (map);
|
||||
}
|
||||
|
||||
static HandleEntry * store_handle (HandleMap *map, dds_instance_handle_t key)
|
||||
{
|
||||
HandleEntry * entry = malloc (sizeof (*entry));
|
||||
assert(entry);
|
||||
memset (entry, 0, sizeof (*entry));
|
||||
|
||||
entry->handle = key;
|
||||
entry->next = map->entries;
|
||||
map->entries = entry;
|
||||
|
||||
return entry;
|
||||
}
|
||||
|
||||
static HandleEntry * retrieve_handle (HandleMap *map, dds_instance_handle_t key)
|
||||
{
|
||||
HandleEntry * entry = map->entries;
|
||||
|
||||
while (entry)
|
||||
{
|
||||
if (entry->handle == key)
|
||||
{
|
||||
break;
|
||||
}
|
||||
entry = entry->next;
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
|
||||
static void data_available_handler (dds_entity_t reader, void *arg)
|
||||
{
|
||||
int samples_received;
|
||||
dds_sample_info_t info [MAX_SAMPLES];
|
||||
dds_instance_handle_t ph = 0;
|
||||
HandleEntry * current = NULL;
|
||||
(void)arg;
|
||||
|
||||
if (startTime == 0)
|
||||
{
|
||||
startTime = dds_time ();
|
||||
}
|
||||
|
||||
/* Take samples and iterate through them */
|
||||
|
||||
samples_received = dds_take (reader, samples, info, MAX_SAMPLES, MAX_SAMPLES);
|
||||
DDS_ERR_CHECK (samples_received, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
for (int i = 0; !done && i < samples_received; i++)
|
||||
{
|
||||
if (info[i].valid_data)
|
||||
{
|
||||
ph = info[i].publication_handle;
|
||||
current = retrieve_handle (imap, ph);
|
||||
ThroughputModule_DataType * this_sample = &data[i];
|
||||
|
||||
if (current == NULL)
|
||||
{
|
||||
current = store_handle (imap, ph);
|
||||
current->count = this_sample->count;
|
||||
}
|
||||
|
||||
if (this_sample->count != current->count)
|
||||
{
|
||||
outOfOrder++;
|
||||
}
|
||||
current->count = this_sample->count + 1;
|
||||
|
||||
/* Add the sample payload size to the total received */
|
||||
|
||||
payloadSize = this_sample->payload._length;
|
||||
total_bytes += payloadSize + 8;
|
||||
total_samples++;
|
||||
}
|
||||
}
|
||||
time_now = dds_time ();
|
||||
if ((pollingDelay == 0) && (time_now > (prev_time + DDS_SECS (1))))
|
||||
{
|
||||
dds_waitset_set_trigger (pollingWaitset, true);
|
||||
}
|
||||
}
|
||||
|
||||
static int parse_args(int argc, char **argv, unsigned long long *maxCycles, char **partitionName)
|
||||
{
|
||||
/*
|
||||
* Get the program parameters
|
||||
* Parameters: subscriber [maxCycles] [pollingDelay] [partitionName]
|
||||
*/
|
||||
if (argc == 2 && (strcmp (argv[1], "-h") == 0 || strcmp (argv[1], "--help") == 0))
|
||||
{
|
||||
printf ("Usage (parameters must be supplied in order):\n");
|
||||
printf ("./subscriber [maxCycles (0 = infinite)] [pollingDelay (ms, 0 = event based)] [partitionName]\n");
|
||||
printf ("Defaults:\n");
|
||||
printf ("./subscriber 0 0 \"Throughput example\"\n");
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
||||
if (argc > 1)
|
||||
{
|
||||
*maxCycles = (unsigned long long) atoi (argv[1]); /* The number of times to output statistics before terminating */
|
||||
}
|
||||
if (argc > 2)
|
||||
{
|
||||
pollingDelay = atoi (argv[2]); /* The number of ms to wait between reads (0 = event based) */
|
||||
}
|
||||
if (argc > 3)
|
||||
{
|
||||
*partitionName = argv[3]; /* The name of the partition */
|
||||
}
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
static void process_samples(unsigned long long maxCycles)
|
||||
{
|
||||
dds_return_t status;
|
||||
unsigned long long prev_bytes = 0;
|
||||
unsigned long long prev_samples = 0;
|
||||
dds_attach_t wsresults[1];
|
||||
size_t wsresultsize = 1U;
|
||||
dds_time_t deltaTv;
|
||||
bool first_batch = true;
|
||||
unsigned long cycles = 0;
|
||||
double deltaTime = 0;
|
||||
|
||||
while (!done && (maxCycles == 0 || cycles < maxCycles))
|
||||
{
|
||||
if (pollingDelay)
|
||||
{
|
||||
dds_sleepfor (DDS_MSECS (pollingDelay));
|
||||
}
|
||||
else
|
||||
{
|
||||
status = dds_waitset_wait (waitSet, wsresults, wsresultsize, DDS_INFINITY);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
if ((status > 0 ) && (dds_triggered (pollingWaitset)))
|
||||
{
|
||||
dds_waitset_set_trigger (pollingWaitset, false);
|
||||
}
|
||||
}
|
||||
|
||||
if (!first_batch)
|
||||
{
|
||||
deltaTv = time_now - prev_time;
|
||||
deltaTime = (double) deltaTv / DDS_NSECS_IN_SEC;
|
||||
prev_time = time_now;
|
||||
|
||||
printf
|
||||
(
|
||||
"=== [Subscriber] Payload size: %lu | Total received: %llu samples, %llu bytes | Out of order: %llu samples "
|
||||
"Transfer rate: %.2lf samples/s, %.2lf Mbit/s\n",
|
||||
payloadSize, total_samples, total_bytes, outOfOrder,
|
||||
(deltaTime != 0.0) ? ((double)(total_samples - prev_samples) / deltaTime) : 0,
|
||||
(deltaTime != 0.0) ? ((double)((total_bytes - prev_bytes) / BYTES_PER_SEC_TO_MEGABITS_PER_SEC) / deltaTime) : 0
|
||||
);
|
||||
cycles++;
|
||||
}
|
||||
else
|
||||
{
|
||||
prev_time = time_now;
|
||||
first_batch = false;
|
||||
}
|
||||
|
||||
/* Update the previous values for next iteration */
|
||||
|
||||
prev_bytes = total_bytes;
|
||||
prev_samples = total_samples;
|
||||
}
|
||||
|
||||
/* Output totals and averages */
|
||||
deltaTv = time_now - startTime;
|
||||
deltaTime = (double) (deltaTv / DDS_NSECS_IN_SEC);
|
||||
printf ("\nTotal received: %llu samples, %llu bytes\n", total_samples, total_bytes);
|
||||
printf ("Out of order: %llu samples\n", outOfOrder);
|
||||
printf ("Average transfer rate: %.2lf samples/s, ", (double)total_samples / deltaTime);
|
||||
printf ("%.2lf Mbit/s\n", (double)(total_bytes / BYTES_PER_SEC_TO_MEGABITS_PER_SEC) / deltaTime);
|
||||
}
|
||||
|
||||
static dds_entity_t prepare_dds(dds_entity_t *reader, const char *partitionName)
|
||||
{
|
||||
dds_return_t status;
|
||||
dds_entity_t topic;
|
||||
dds_entity_t subscriber;
|
||||
dds_listener_t *rd_listener;
|
||||
dds_entity_t participant;
|
||||
|
||||
int32_t maxSamples = 400;
|
||||
const char *subParts[1];
|
||||
dds_qos_t *subQos = dds_qos_create ();
|
||||
dds_qos_t *drQos = dds_qos_create ();
|
||||
|
||||
/* A Participant is created for the default domain. */
|
||||
|
||||
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
|
||||
DDS_ERR_CHECK (participant, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A Topic is created for our sample type on the domain participant. */
|
||||
|
||||
topic = dds_create_topic (participant, &ThroughputModule_DataType_desc, "Throughput", NULL, NULL);
|
||||
DDS_ERR_CHECK (topic, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
/* A Subscriber is created on the domain participant. */
|
||||
|
||||
subParts[0] = partitionName;
|
||||
dds_qset_partition (subQos, 1, subParts);
|
||||
subscriber = dds_create_subscriber (participant, subQos, NULL);
|
||||
DDS_ERR_CHECK (subscriber, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (subQos);
|
||||
|
||||
/* A Reader is created on the Subscriber & Topic with a modified Qos. */
|
||||
|
||||
dds_qset_reliability (drQos, DDS_RELIABILITY_RELIABLE, DDS_SECS (10));
|
||||
dds_qset_history (drQos, DDS_HISTORY_KEEP_ALL, 0);
|
||||
dds_qset_resource_limits (drQos, maxSamples, DDS_LENGTH_UNLIMITED, DDS_LENGTH_UNLIMITED);
|
||||
|
||||
rd_listener = dds_listener_create(NULL);
|
||||
dds_lset_data_available(rd_listener, data_available_handler);
|
||||
|
||||
/* A Read Condition is created which is triggered when data is available to read */
|
||||
waitSet = dds_create_waitset (participant);
|
||||
DDS_ERR_CHECK (waitSet, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
pollingWaitset = dds_create_waitset (participant);
|
||||
DDS_ERR_CHECK (pollingWaitset, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
status = dds_waitset_attach (waitSet, pollingWaitset, pollingWaitset);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
status = dds_waitset_attach (waitSet, waitSet, waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
|
||||
imap = HandleMap__alloc ();
|
||||
|
||||
memset (data, 0, sizeof (data));
|
||||
for (unsigned int i = 0; i < MAX_SAMPLES; i++)
|
||||
{
|
||||
samples[i] = &data[i];
|
||||
}
|
||||
|
||||
*reader = dds_create_reader (subscriber, topic, drQos, rd_listener);
|
||||
DDS_ERR_CHECK (*reader, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
dds_qos_delete (drQos);
|
||||
dds_listener_delete(rd_listener);
|
||||
|
||||
return participant;
|
||||
}
|
||||
|
||||
static void finalize_dds(dds_entity_t participant)
|
||||
{
|
||||
dds_return_t status;
|
||||
|
||||
for (unsigned int i = 0; i < MAX_SAMPLES; i++)
|
||||
{
|
||||
ThroughputModule_DataType_free (&data[i], DDS_FREE_CONTENTS);
|
||||
}
|
||||
|
||||
status = dds_waitset_detach (waitSet, waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_waitset_detach (waitSet, pollingWaitset);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_delete (pollingWaitset);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_delete (waitSet);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
status = dds_delete (participant);
|
||||
DDS_ERR_CHECK (status, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue