cyclonedds/src/core/ddsc/tests/liveliness.c

615 lines
25 KiB
C
Raw Normal View History

Liveliness QoS implementation for auto and manual-by-participant This commit adds support for the liveliness QoS for the liveliness kinds automatic and manual-by-participant. It also implements the lease_duration from this QoS, which was ignored until now. In the api the function dds_assert_liveliness is added to assert liveliness on a participant, which can be used when using liveliness kind manual-by-participant. Liveliness kind manual-by-topic is not yet supported, this will be added in a later commit. * Proxy participants now have 2 fibheaps to keep leases: one for leases of pwrs with automatic liveliness and one for leases of the pwrs with manual-by-participant liveliness (both protected by the proxypp lock). The minl_auto and minl_man members represent the shortest lease from these fibheaps and these leases are renewed when receiving data. Replacing the minl_ leases is now done by replacing the lease object (atomic ptr) with delayed deletion of the old lease using the gc. * Proxy writers are set not-alive when the lease expired, and reset to alive then data is received. When data is received by a pwr, the other pwrs in the proxypp might also be set alive. I think the specification is not clear at this point, and for now I have not implemented this * I refactored out the counter for man-by-pp proxy writers and improved locking when updating the min-leases on the proxy participant, so I think this fixes the race conditions. Some additional tests are required, e.g. to test the not-alive->alive transition for pwrs. I will add these in short term, as well as the implementation of the manual-by-topic liveliness kind. Signed-off-by: Dennis Potman <dennis.potman@adlinktech.com>
2019-10-09 10:28:19 +02:00
/*
* Copyright(c) 2006 to 2018 ADLINK Technology Limited and others
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v. 2.0 which is available at
* http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License
* v. 1.0 which is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
*/
#include <assert.h>
#include <limits.h>
#include "dds/dds.h"
#include "CUnit/Theory.h"
#include "Space.h"
#include "config_env.h"
#include "dds/version.h"
#include "dds__entity.h"
#include "dds/ddsi/q_entity.h"
#include "dds/ddsrt/cdtors.h"
#include "dds/ddsrt/misc.h"
#include "dds/ddsrt/process.h"
#include "dds/ddsrt/threads.h"
#include "dds/ddsrt/environ.h"
#include "dds/ddsrt/atomics.h"
#include "dds/ddsrt/time.h"
#define DDS_DOMAINID_PUB 0
#define DDS_DOMAINID_SUB 1
#define DDS_CONFIG_NO_PORT_GAIN "${CYCLONEDDS_URI}${CYCLONEDDS_URI:+,}<General><NetworkInterfaceAddress>127.0.0.1</NetworkInterfaceAddress></General><Discovery><Ports><DomainGain>0</DomainGain></Ports></Discovery>"
#define DDS_CONFIG_NO_PORT_GAIN_LOG "<"DDS_PROJECT_NAME"><Domain><Discovery><Ports><DomainGain>0</DomainGain></Ports></Discovery><Tracing><OutputFile>cyclonedds_liveliness.log</OutputFile><Verbosity>finest</Verbosity></Tracing></Domain></"DDS_PROJECT_NAME">"
Liveliness QoS implementation for auto and manual-by-participant This commit adds support for the liveliness QoS for the liveliness kinds automatic and manual-by-participant. It also implements the lease_duration from this QoS, which was ignored until now. In the api the function dds_assert_liveliness is added to assert liveliness on a participant, which can be used when using liveliness kind manual-by-participant. Liveliness kind manual-by-topic is not yet supported, this will be added in a later commit. * Proxy participants now have 2 fibheaps to keep leases: one for leases of pwrs with automatic liveliness and one for leases of the pwrs with manual-by-participant liveliness (both protected by the proxypp lock). The minl_auto and minl_man members represent the shortest lease from these fibheaps and these leases are renewed when receiving data. Replacing the minl_ leases is now done by replacing the lease object (atomic ptr) with delayed deletion of the old lease using the gc. * Proxy writers are set not-alive when the lease expired, and reset to alive then data is received. When data is received by a pwr, the other pwrs in the proxypp might also be set alive. I think the specification is not clear at this point, and for now I have not implemented this * I refactored out the counter for man-by-pp proxy writers and improved locking when updating the min-leases on the proxy participant, so I think this fixes the race conditions. Some additional tests are required, e.g. to test the not-alive->alive transition for pwrs. I will add these in short term, as well as the implementation of the manual-by-topic liveliness kind. Signed-off-by: Dennis Potman <dennis.potman@adlinktech.com>
2019-10-09 10:28:19 +02:00
uint32_t g_topic_nr = 0;
static dds_entity_t g_pub_domain = 0;
static dds_entity_t g_pub_participant = 0;
static dds_entity_t g_pub_publisher = 0;
static dds_entity_t g_sub_domain = 0;
static dds_entity_t g_sub_participant = 0;
static dds_entity_t g_sub_subscriber = 0;
static char *create_topic_name(const char *prefix, uint32_t nr, char *name, size_t size)
{
/* Get semi random g_topic name. */
ddsrt_pid_t pid = ddsrt_getpid();
ddsrt_tid_t tid = ddsrt_gettid();
(void)snprintf(name, size, "%s%d_pid%" PRIdPID "_tid%" PRIdTID "", prefix, nr, pid, tid);
return name;
}
static void liveliness_init(void)
{
char *conf = ddsrt_expand_envvars(DDS_CONFIG_NO_PORT_GAIN, UINT32_MAX);
g_pub_domain = dds_create_domain(DDS_DOMAINID_PUB, conf);
g_sub_domain = dds_create_domain(DDS_DOMAINID_SUB, conf);
dds_free(conf);
g_pub_participant = dds_create_participant(DDS_DOMAINID_PUB, NULL, NULL);
CU_ASSERT_FATAL(g_pub_participant > 0);
g_sub_participant = dds_create_participant(DDS_DOMAINID_SUB, NULL, NULL);
CU_ASSERT_FATAL(g_sub_participant > 0);
g_pub_publisher = dds_create_publisher(g_pub_participant, NULL, NULL);
CU_ASSERT_FATAL(g_pub_publisher > 0);
g_sub_subscriber = dds_create_subscriber(g_sub_participant, NULL, NULL);
CU_ASSERT_FATAL(g_sub_subscriber > 0);
}
static void liveliness_fini(void)
{
dds_delete(g_sub_subscriber);
dds_delete(g_pub_publisher);
dds_delete(g_sub_participant);
dds_delete(g_pub_participant);
dds_delete(g_sub_domain);
dds_delete(g_pub_domain);
}
static seqno_t get_pmd_seqno(dds_entity_t participant)
{
seqno_t seqno;
struct dds_entity *pp_entity;
struct participant *pp;
struct writer *wr;
CU_ASSERT_EQUAL_FATAL(dds_entity_pin(participant, &pp_entity), 0);
thread_state_awake(lookup_thread_state(), &pp_entity->m_domain->gv);
pp = ephash_lookup_participant_guid(pp_entity->m_domain->gv.guid_hash, &pp_entity->m_guid);
wr = get_builtin_writer(pp, NN_ENTITYID_P2P_BUILTIN_PARTICIPANT_MESSAGE_WRITER);
CU_ASSERT_FATAL(wr != NULL);
seqno = wr->seq;
thread_state_asleep(lookup_thread_state());
dds_entity_unpin(pp_entity);
return seqno;
}
static dds_duration_t get_pmd_interval(dds_entity_t participant)
{
dds_duration_t intv;
struct dds_entity *pp_entity;
struct participant *pp;
CU_ASSERT_EQUAL_FATAL(dds_entity_pin(participant, &pp_entity), 0);
thread_state_awake(lookup_thread_state(), &pp_entity->m_domain->gv);
pp = ephash_lookup_participant_guid(pp_entity->m_domain->gv.guid_hash, &pp_entity->m_guid);
intv = pp_get_pmd_interval(pp);
thread_state_asleep(lookup_thread_state());
dds_entity_unpin(pp_entity);
return intv;
}
static dds_duration_t get_ldur_config(dds_entity_t participant)
{
struct dds_entity *pp_entity;
dds_duration_t ldur;
CU_ASSERT_EQUAL_FATAL(dds_entity_pin(participant, &pp_entity), 0);
ldur = (dds_duration_t)pp_entity->m_domain->gv.config.lease_duration;
dds_entity_unpin(pp_entity);
return ldur;
}
#define A DDS_LIVELINESS_AUTOMATIC
#define MP DDS_LIVELINESS_MANUAL_BY_PARTICIPANT
CU_TheoryDataPoints(ddsc_liveliness, pmd_count) = {
CU_DataPoints(dds_liveliness_kind_t, A, A, MP), /* liveliness kind */
CU_DataPoints(uint32_t, 200, 200, 200), /* lease duration */
CU_DataPoints(double, 5, 10, 5), /* delay (n times lease duration) */
};
#undef A
#undef MP
CU_Theory((dds_liveliness_kind_t kind, uint32_t ldur, double mult), ddsc_liveliness, pmd_count, .init = liveliness_init, .fini = liveliness_fini, .timeout = 30)
Liveliness QoS implementation for auto and manual-by-participant This commit adds support for the liveliness QoS for the liveliness kinds automatic and manual-by-participant. It also implements the lease_duration from this QoS, which was ignored until now. In the api the function dds_assert_liveliness is added to assert liveliness on a participant, which can be used when using liveliness kind manual-by-participant. Liveliness kind manual-by-topic is not yet supported, this will be added in a later commit. * Proxy participants now have 2 fibheaps to keep leases: one for leases of pwrs with automatic liveliness and one for leases of the pwrs with manual-by-participant liveliness (both protected by the proxypp lock). The minl_auto and minl_man members represent the shortest lease from these fibheaps and these leases are renewed when receiving data. Replacing the minl_ leases is now done by replacing the lease object (atomic ptr) with delayed deletion of the old lease using the gc. * Proxy writers are set not-alive when the lease expired, and reset to alive then data is received. When data is received by a pwr, the other pwrs in the proxypp might also be set alive. I think the specification is not clear at this point, and for now I have not implemented this * I refactored out the counter for man-by-pp proxy writers and improved locking when updating the min-leases on the proxy participant, so I think this fixes the race conditions. Some additional tests are required, e.g. to test the not-alive->alive transition for pwrs. I will add these in short term, as well as the implementation of the manual-by-topic liveliness kind. Signed-off-by: Dennis Potman <dennis.potman@adlinktech.com>
2019-10-09 10:28:19 +02:00
{
dds_entity_t pub_topic;
dds_entity_t sub_topic;
dds_entity_t reader;
dds_entity_t writer;
seqno_t start_seqno, end_seqno;
dds_qos_t *rqos;
dds_qos_t *wqos;
dds_entity_t waitset;
dds_attach_t triggered;
uint32_t status;
char name[100];
dds_time_t t;
t = dds_time();
printf("%d.%06d running test: kind %s, lease duration %d, delay %f\n",
(int32_t)(t / DDS_NSECS_IN_SEC), (int32_t)(t % DDS_NSECS_IN_SEC) / 1000,
kind == 0 ? "A" : "MP", ldur, mult);
/* topics */
create_topic_name("ddsc_liveliness_test", g_topic_nr++, name, sizeof name);
CU_ASSERT_FATAL((pub_topic = dds_create_topic(g_pub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
CU_ASSERT_FATAL((sub_topic = dds_create_topic(g_sub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
/* reader */
CU_ASSERT_FATAL((rqos = dds_create_qos()) != NULL);
dds_qset_liveliness(rqos, DDS_LIVELINESS_AUTOMATIC, DDS_INFINITY);
CU_ASSERT_FATAL((reader = dds_create_reader(g_sub_participant, sub_topic, rqos, NULL)) > 0);
dds_delete_qos(rqos);
CU_ASSERT_EQUAL_FATAL(dds_set_status_mask(reader, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* waitset on reader */
CU_ASSERT_FATAL((waitset = dds_create_waitset(g_sub_participant)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_attach(waitset, reader, reader), DDS_RETCODE_OK);
/* writer */
CU_ASSERT_FATAL((wqos = dds_create_qos()) != NULL);
dds_qset_liveliness(wqos, kind, DDS_MSECS(ldur));
CU_ASSERT_FATAL((writer = dds_create_writer(g_pub_participant, pub_topic, wqos, NULL)) > 0);
dds_delete_qos(wqos);
/* wait for writer to be alive */
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_SECS(1)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* check no of PMD messages sent */
start_seqno = get_pmd_seqno(g_pub_participant);
dds_sleepfor(DDS_MSECS((dds_duration_t)(mult * ldur)));
end_seqno = get_pmd_seqno(g_pub_participant);
t = dds_time();
printf("%d.%06d PMD sequence no: start %" PRId64 " -> end %" PRId64 "\n",
(int32_t)(t / DDS_NSECS_IN_SEC), (int32_t)(t % DDS_NSECS_IN_SEC) / 1000,
start_seqno, end_seqno);
/* end-start should be mult - 1, but allow 1 pmd sample to be lost */
CU_ASSERT(end_seqno - start_seqno >= (kind == DDS_LIVELINESS_AUTOMATIC ? mult - 2 : 0))
Liveliness QoS implementation for auto and manual-by-participant This commit adds support for the liveliness QoS for the liveliness kinds automatic and manual-by-participant. It also implements the lease_duration from this QoS, which was ignored until now. In the api the function dds_assert_liveliness is added to assert liveliness on a participant, which can be used when using liveliness kind manual-by-participant. Liveliness kind manual-by-topic is not yet supported, this will be added in a later commit. * Proxy participants now have 2 fibheaps to keep leases: one for leases of pwrs with automatic liveliness and one for leases of the pwrs with manual-by-participant liveliness (both protected by the proxypp lock). The minl_auto and minl_man members represent the shortest lease from these fibheaps and these leases are renewed when receiving data. Replacing the minl_ leases is now done by replacing the lease object (atomic ptr) with delayed deletion of the old lease using the gc. * Proxy writers are set not-alive when the lease expired, and reset to alive then data is received. When data is received by a pwr, the other pwrs in the proxypp might also be set alive. I think the specification is not clear at this point, and for now I have not implemented this * I refactored out the counter for man-by-pp proxy writers and improved locking when updating the min-leases on the proxy participant, so I think this fixes the race conditions. Some additional tests are required, e.g. to test the not-alive->alive transition for pwrs. I will add these in short term, as well as the implementation of the manual-by-topic liveliness kind. Signed-off-by: Dennis Potman <dennis.potman@adlinktech.com>
2019-10-09 10:28:19 +02:00
if (kind == DDS_LIVELINESS_MANUAL_BY_PARTICIPANT)
CU_ASSERT(get_pmd_seqno(g_pub_participant) - start_seqno < mult)
/* cleanup */
CU_ASSERT_EQUAL_FATAL(dds_delete(sub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(pub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(writer), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(reader), DDS_RETCODE_OK);
}
/* FIXME: add DDS_LIVELINESS_MANUAL_BY_TOPIC */
CU_TheoryDataPoints(ddsc_liveliness, expire_liveliness_kinds) = {
CU_DataPoints(uint32_t, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200), /* lease duration */
CU_DataPoints(double, 0.3, 0.3, 0.3, 0.3, 2, 2, 2, 2, 2, 2, 2, 2, 2), /* delay (n times lease duration) */
CU_DataPoints(size_t, 1, 0, 2, 0, 1, 0, 1, 2, 0, 5, 0, 15, 15), /* number of writers with automatic liveliness */
CU_DataPoints(size_t, 1, 1, 2, 2, 1, 1, 0, 2, 2, 5, 10, 0, 15), /* number of writers with manual-by-participant liveliness */
};
CU_Theory((uint32_t ldur, double mult, size_t wr_cnt_auto, size_t wr_cnt_man_pp), ddsc_liveliness, expire_liveliness_kinds, .init = liveliness_init, .fini = liveliness_fini, .timeout = 60)
{
dds_entity_t pub_topic;
dds_entity_t sub_topic;
dds_entity_t reader;
dds_entity_t *writers;
dds_qos_t *rqos, *wqos_auto, *wqos_man_pp;
dds_entity_t waitset;
dds_attach_t triggered;
struct dds_liveliness_changed_status lstatus;
uint32_t status;
size_t n, run = 1;
char name[100];
size_t wr_cnt = wr_cnt_auto + wr_cnt_man_pp;
dds_time_t tstart, t;
bool test_finished = false;
do
{
tstart = dds_time();
printf("%d.%06d running test: lease duration %d, delay %f, auto/manual-by-participant %zu/%zu\n",
(int32_t)(tstart / DDS_NSECS_IN_SEC), (int32_t)(tstart % DDS_NSECS_IN_SEC) / 1000,
ldur, mult, wr_cnt_auto, wr_cnt_man_pp);
/* topics */
create_topic_name("ddsc_liveliness_test", g_topic_nr++, name, sizeof name);
CU_ASSERT_FATAL((pub_topic = dds_create_topic(g_pub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
CU_ASSERT_FATAL((sub_topic = dds_create_topic(g_sub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
/* reader */
CU_ASSERT_FATAL((rqos = dds_create_qos()) != NULL);
dds_qset_liveliness(rqos, DDS_LIVELINESS_AUTOMATIC, DDS_INFINITY);
CU_ASSERT_FATAL((reader = dds_create_reader(g_sub_participant, sub_topic, rqos, NULL)) > 0);
dds_delete_qos(rqos);
CU_ASSERT_EQUAL_FATAL(dds_set_status_mask(reader, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* writers */
CU_ASSERT_FATAL((wqos_auto = dds_create_qos()) != NULL);
dds_qset_liveliness(wqos_auto, DDS_LIVELINESS_AUTOMATIC, DDS_MSECS(ldur));
CU_ASSERT_FATAL((wqos_man_pp = dds_create_qos()) != NULL);
dds_qset_liveliness(wqos_man_pp, DDS_LIVELINESS_MANUAL_BY_PARTICIPANT, DDS_MSECS(ldur));
CU_ASSERT_FATAL((waitset = dds_create_waitset(g_sub_participant)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_attach(waitset, reader, reader), DDS_RETCODE_OK);
writers = dds_alloc(wr_cnt * sizeof(dds_entity_t));
for (n = 0; n < wr_cnt; n++)
{
CU_ASSERT_FATAL((writers[n] = dds_create_writer(g_pub_participant, pub_topic, n < wr_cnt_auto ? wqos_auto : wqos_man_pp, NULL)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_SECS(5)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
}
dds_delete_qos(wqos_auto);
dds_delete_qos(wqos_man_pp);
t = dds_time();
if (t - tstart > DDS_MSECS(0.5 * ldur))
{
ldur *= 10;
printf("%d.%06d failed to create writers in time\n",
(int32_t)(t / DDS_NSECS_IN_SEC), (int32_t)(t % DDS_NSECS_IN_SEC) / 1000);
}
else
{
/* check alive count before proxy writers are expired */
dds_get_liveliness_changed_status(reader, &lstatus);
printf("%d.%06d writers alive: %d\n", (int32_t)(t / DDS_NSECS_IN_SEC), (int32_t)(t % DDS_NSECS_IN_SEC) / 1000, lstatus.alive_count);
CU_ASSERT_EQUAL_FATAL(lstatus.alive_count, wr_cnt);
dds_time_t tstop = tstart + DDS_MSECS((dds_duration_t)(mult * ldur));
size_t stopped = 0;
do
{
dds_duration_t w = tstop - dds_time();
CU_ASSERT_FATAL((dds_waitset_wait(waitset, &triggered, 1, w > 0 ? w : 0)) >= 0);
CU_ASSERT_EQUAL_FATAL(dds_get_liveliness_changed_status(reader, &lstatus), DDS_RETCODE_OK);
stopped += (uint32_t)lstatus.not_alive_count_change;
} while (dds_time() < tstop);
t = dds_time();
printf("%d.%06d writers stopped: %zu\n",
(int32_t)(t / DDS_NSECS_IN_SEC), (int32_t)(t % DDS_NSECS_IN_SEC) / 1000, stopped);
size_t exp_stopped = mult < 1 ? 0 : wr_cnt_man_pp;
if (stopped != exp_stopped)
{
ldur *= 10;
printf("%d.%06d incorrect number of stopped writers\n",
(int32_t)(t / DDS_NSECS_IN_SEC), (int32_t)(t % DDS_NSECS_IN_SEC) / 1000);
}
else
{
/* check alive count */
CU_ASSERT_EQUAL_FATAL(dds_get_liveliness_changed_status(reader, &lstatus), DDS_RETCODE_OK);
CU_ASSERT_EQUAL(lstatus.alive_count, mult < 1 ? wr_cnt : wr_cnt_auto);
test_finished = true;
}
}
/* cleanup */
CU_ASSERT_EQUAL_FATAL(dds_waitset_detach(waitset, reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(waitset), DDS_RETCODE_OK);
for (n = 0; n < wr_cnt; n++)
CU_ASSERT_EQUAL_FATAL(dds_delete(writers[n]), DDS_RETCODE_OK);
dds_free(writers);
CU_ASSERT_EQUAL_FATAL(dds_delete(sub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(pub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(reader), DDS_RETCODE_OK);
if (!test_finished)
{
if (run++ > 2)
{
printf("%d.%06d run limit reached, test failed\n", (int32_t)(tstart / DDS_NSECS_IN_SEC), (int32_t)(tstart % DDS_NSECS_IN_SEC) / 1000);
CU_FAIL_FATAL("Run limit reached");
test_finished = true;
continue;
}
else
{
printf("%d.%06d restarting test with ldur %d\n",
(int32_t)(t / DDS_NSECS_IN_SEC), (int32_t)(t % DDS_NSECS_IN_SEC) / 1000, ldur);
}
}
} while (!test_finished);
}
static void add_and_check_writer(dds_liveliness_kind_t kind, dds_duration_t ldur, dds_entity_t *writer, dds_entity_t topic, dds_entity_t reader)
{
dds_entity_t waitset;
dds_qos_t *wqos;
dds_attach_t triggered;
uint32_t status;
CU_ASSERT_FATAL((waitset = dds_create_waitset(g_sub_participant)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_attach(waitset, reader, reader), DDS_RETCODE_OK);
CU_ASSERT_FATAL((wqos = dds_create_qos()) != NULL);
dds_qset_liveliness(wqos, kind, ldur);
CU_ASSERT_FATAL((*writer = dds_create_writer(g_pub_participant, topic, wqos, NULL)) > 0);
dds_delete_qos(wqos);
/* wait for writer to be alive */
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_MSECS(1000)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_waitset_detach(waitset, reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(waitset), DDS_RETCODE_OK);
}
#define MAX_WRITERS 10
CU_Test(ddsc_liveliness, lease_duration, .init = liveliness_init, .fini = liveliness_fini)
{
dds_entity_t pub_topic;
dds_entity_t sub_topic;
dds_entity_t reader;
dds_entity_t writers[MAX_WRITERS];
size_t wr_cnt = 0;
char name[100];
dds_qos_t *rqos;
uint32_t n;
/* topics */
create_topic_name("ddsc_liveliness_ldur", 1, name, sizeof name);
CU_ASSERT_FATAL((pub_topic = dds_create_topic(g_pub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
CU_ASSERT_FATAL((sub_topic = dds_create_topic(g_sub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
/* reader and waitset */
CU_ASSERT_FATAL((rqos = dds_create_qos()) != NULL);
dds_qset_liveliness(rqos, DDS_LIVELINESS_AUTOMATIC, DDS_INFINITY);
CU_ASSERT_FATAL((reader = dds_create_reader(g_sub_participant, sub_topic, rqos, NULL)) > 0);
dds_delete_qos(rqos);
CU_ASSERT_EQUAL_FATAL(dds_set_status_mask(reader, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* check if pmd defaults to configured duration */
CU_ASSERT_EQUAL_FATAL(get_pmd_interval(g_pub_participant), get_ldur_config(g_pub_participant));
/* create writers and check pmd interval in publishing participant */
add_and_check_writer(DDS_LIVELINESS_AUTOMATIC, DDS_MSECS(1000), &writers[wr_cnt++], pub_topic, reader);
CU_ASSERT_EQUAL_FATAL(get_pmd_interval(g_pub_participant), DDS_MSECS(1000));
add_and_check_writer(DDS_LIVELINESS_AUTOMATIC, DDS_MSECS(2000), &writers[wr_cnt++], pub_topic, reader);
CU_ASSERT_EQUAL_FATAL(get_pmd_interval(g_pub_participant), DDS_MSECS(1000));
add_and_check_writer(DDS_LIVELINESS_MANUAL_BY_PARTICIPANT, DDS_MSECS(2000), &writers[wr_cnt++], pub_topic, reader);
CU_ASSERT_EQUAL_FATAL(get_pmd_interval(g_pub_participant), DDS_MSECS(1000));
add_and_check_writer(DDS_LIVELINESS_AUTOMATIC, DDS_MSECS(500), &writers[wr_cnt++], pub_topic, reader);
CU_ASSERT_EQUAL_FATAL(get_pmd_interval(g_pub_participant), DDS_MSECS(500));
add_and_check_writer(DDS_LIVELINESS_MANUAL_BY_PARTICIPANT, DDS_MSECS(100), &writers[wr_cnt++], pub_topic, reader);
CU_ASSERT_EQUAL_FATAL(get_pmd_interval(g_pub_participant), DDS_MSECS(500));
/* cleanup */
for (n = 0; n < wr_cnt; n++)
CU_ASSERT_EQUAL_FATAL(dds_delete(writers[n]), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(sub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(pub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(reader), DDS_RETCODE_OK);
}
#undef MAX_WRITERS
CU_Test(ddsc_liveliness, lease_duration_pwr, .init = liveliness_init, .fini = liveliness_fini)
{
dds_entity_t pub_topic;
dds_entity_t sub_topic;
dds_entity_t reader;
dds_entity_t writer;
char name[100];
dds_qos_t *rqos, *wqos;
dds_entity_t waitset;
dds_attach_t triggered;
uint32_t status;
dds_duration_t ldur;
/* topics */
create_topic_name("ddsc_liveliness_ldurpwr", 1, name, sizeof name);
CU_ASSERT_FATAL((pub_topic = dds_create_topic(g_pub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
CU_ASSERT_FATAL((sub_topic = dds_create_topic(g_sub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
/* reader */
CU_ASSERT_FATAL((rqos = dds_create_qos()) != NULL);
dds_qset_liveliness(rqos, DDS_LIVELINESS_AUTOMATIC, DDS_INFINITY);
CU_ASSERT_FATAL((reader = dds_create_reader(g_sub_participant, sub_topic, rqos, NULL)) > 0);
dds_delete_qos(rqos);
CU_ASSERT_EQUAL_FATAL(dds_set_status_mask(reader, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* writer */
ldur = 1000;
CU_ASSERT_FATAL((wqos = dds_create_qos()) != NULL);
dds_qset_liveliness(wqos, DDS_LIVELINESS_AUTOMATIC, DDS_MSECS(ldur));
CU_ASSERT_FATAL((writer = dds_create_writer(g_pub_participant, pub_topic, wqos, NULL)) > 0);
/* wait for writer to be alive */
CU_ASSERT_FATAL((waitset = dds_create_waitset(g_sub_participant)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_attach(waitset, reader, reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_MSECS(1000)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* check pwr lease duration in matched publication */
dds_instance_handle_t wrs[1];
CU_ASSERT_EQUAL_FATAL(dds_get_matched_publications(reader, wrs, 1), 1);
dds_builtintopic_endpoint_t *ep;
ep = dds_get_matched_publication_data(reader, wrs[0]);
CU_ASSERT_FATAL(ep != NULL);
CU_ASSERT_EQUAL_FATAL(ep->qos->liveliness.lease_duration, DDS_MSECS(ldur));
dds_delete_qos(ep->qos);
dds_free(ep->topic_name);
dds_free(ep->type_name);
dds_free(ep);
/* cleanup */
dds_delete_qos(wqos);
CU_ASSERT_EQUAL_FATAL(dds_waitset_detach(waitset, reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(waitset), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(writer), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(sub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(pub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(reader), DDS_RETCODE_OK);
}
#define MAX_WRITERS 100
CU_Test(ddsc_liveliness, create_delete_writer_stress, .init = liveliness_init, .fini = liveliness_fini)
{
dds_entity_t pub_topic;
dds_entity_t sub_topic;
dds_entity_t reader;
dds_entity_t writers[MAX_WRITERS];
dds_entity_t waitset;
dds_qos_t *wqos;
struct dds_liveliness_changed_status lstatus;
size_t wr_cnt = 0;
char name[100];
dds_qos_t *rqos;
dds_attach_t triggered;
uint32_t n, status;
Space_Type1 sample = { 0, 0, 0 };
/* topics */
create_topic_name("ddsc_liveliness_wr_stress", 1, name, sizeof name);
CU_ASSERT_FATAL((pub_topic = dds_create_topic(g_pub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
CU_ASSERT_FATAL((sub_topic = dds_create_topic(g_sub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
/* reader and waitset */
CU_ASSERT_FATAL((rqos = dds_create_qos()) != NULL);
dds_qset_liveliness(rqos, DDS_LIVELINESS_AUTOMATIC, DDS_INFINITY);
CU_ASSERT_FATAL((reader = dds_create_reader(g_sub_participant, sub_topic, rqos, NULL)) > 0);
dds_delete_qos(rqos);
CU_ASSERT_EQUAL_FATAL(dds_set_status_mask(reader, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
CU_ASSERT_FATAL((waitset = dds_create_waitset(g_sub_participant)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_attach(waitset, reader, reader), DDS_RETCODE_OK);
/* create 1st writer and wait for it to become alive */
CU_ASSERT_FATAL((wqos = dds_create_qos()) != NULL);
dds_qset_liveliness(wqos, DDS_LIVELINESS_MANUAL_BY_PARTICIPANT, DDS_MSECS (500));
CU_ASSERT_FATAL((writers[0] = dds_create_writer(g_pub_participant, pub_topic, wqos, NULL)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_MSECS(1000)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* create writers */
for (n = 1; n < MAX_WRITERS; n++)
{
dds_qset_liveliness(wqos, n % 1 ? DDS_LIVELINESS_AUTOMATIC : DDS_LIVELINESS_MANUAL_BY_PARTICIPANT, DDS_MSECS (500 - n));
CU_ASSERT_FATAL((writers[n] = dds_create_writer(g_pub_participant, pub_topic, wqos, NULL)) > 0);
dds_write (writers[n], &sample);
if (n % 3 == 2)
dds_delete(writers[n]);
}
dds_delete_qos(wqos);
/* wait for all writers to become alive */
while (wr_cnt < MAX_WRITERS)
{
CU_ASSERT_EQUAL_FATAL(dds_get_liveliness_changed_status (reader, &lstatus), DDS_RETCODE_OK);
wr_cnt += (uint32_t)lstatus.alive_count_change;
dds_sleepfor (DDS_MSECS(50));
}
CU_ASSERT_EQUAL_FATAL (wr_cnt, MAX_WRITERS);
/* cleanup remaining writers */
for (n = 0; n < wr_cnt; n++)
{
if (n % 3 != 2)
CU_ASSERT_EQUAL_FATAL(dds_delete(writers[n]), DDS_RETCODE_OK);
}
CU_ASSERT_EQUAL_FATAL(dds_waitset_detach(waitset, reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(waitset), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(sub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(pub_topic), DDS_RETCODE_OK);
}
#undef MAX_WRITERS
CU_Test(ddsc_liveliness, status_counts, .init = liveliness_init, .fini = liveliness_fini)
{
dds_entity_t pub_topic;
dds_entity_t sub_topic;
dds_entity_t reader;
dds_entity_t writer;
dds_entity_t waitset;
dds_qos_t *rqos;
dds_qos_t *wqos;
dds_attach_t triggered;
uint32_t status = 0;
struct dds_liveliness_changed_status lstatus;
struct dds_subscription_matched_status sstatus;
char name[100];
dds_duration_t ldur = DDS_MSECS (500);
Space_Type1 sample = { 1, 0, 0 };
/* topics */
create_topic_name("ddsc_liveliness_status_counts", g_topic_nr++, name, sizeof name);
CU_ASSERT_FATAL((pub_topic = dds_create_topic(g_pub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
CU_ASSERT_FATAL((sub_topic = dds_create_topic(g_sub_participant, &Space_Type1_desc, name, NULL, NULL)) > 0);
/* reader */
CU_ASSERT_FATAL((rqos = dds_create_qos()) != NULL);
dds_qset_liveliness(rqos, DDS_LIVELINESS_AUTOMATIC, DDS_INFINITY);
CU_ASSERT_FATAL((reader = dds_create_reader(g_sub_participant, sub_topic, rqos, NULL)) > 0);
dds_delete_qos(rqos);
CU_ASSERT_EQUAL_FATAL(dds_set_status_mask(reader, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
CU_ASSERT_FATAL((waitset = dds_create_waitset(g_sub_participant)) > 0);
CU_ASSERT_EQUAL_FATAL(dds_waitset_attach(waitset, reader, reader), DDS_RETCODE_OK);
/* writer */
CU_ASSERT_FATAL((wqos = dds_create_qos()) != NULL);
dds_qset_liveliness(wqos, DDS_LIVELINESS_MANUAL_BY_PARTICIPANT, ldur);
CU_ASSERT_FATAL((writer = dds_create_writer(g_pub_participant, pub_topic, wqos, NULL)) > 0);
dds_delete_qos(wqos);
/* wait for writer to be alive */
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_SECS(5)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
/* check status counts before proxy writer is expired */
dds_get_liveliness_changed_status(reader, &lstatus);
CU_ASSERT_EQUAL_FATAL(lstatus.alive_count, 1);
dds_get_subscription_matched_status(reader, &sstatus);
CU_ASSERT_EQUAL_FATAL(sstatus.current_count, 1);
/* sleep for more than lease duration, writer should be set not-alive but subscription still matched */
dds_sleepfor(ldur + DDS_MSECS(100));
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_SECS(5)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
dds_get_liveliness_changed_status(reader, &lstatus);
CU_ASSERT_EQUAL_FATAL(lstatus.alive_count, 0);
dds_get_subscription_matched_status(reader, &sstatus);
CU_ASSERT_EQUAL_FATAL(sstatus.current_count, 1);
/* write sample and re-check status counts */
dds_write (writer, &sample);
CU_ASSERT_EQUAL_FATAL(dds_waitset_wait(waitset, &triggered, 1, DDS_SECS(5)), 1);
CU_ASSERT_EQUAL_FATAL(dds_take_status(reader, &status, DDS_LIVELINESS_CHANGED_STATUS), DDS_RETCODE_OK);
dds_get_liveliness_changed_status(reader, &lstatus);
CU_ASSERT_EQUAL_FATAL(lstatus.alive_count, 1);
dds_get_subscription_matched_status(reader, &sstatus);
CU_ASSERT_EQUAL_FATAL(sstatus.current_count, 1);
/* cleanup */
CU_ASSERT_EQUAL_FATAL(dds_waitset_detach(waitset, reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(waitset), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(reader), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(writer), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(sub_topic), DDS_RETCODE_OK);
CU_ASSERT_EQUAL_FATAL(dds_delete(pub_topic), DDS_RETCODE_OK);
}