niklas/cyclonedds
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #111 from eboasson/fix-87

Browse source 
Add support for query conditions
This commit is contained in:
eboasson 2019-02-13 09:37:04 +00:00 • committed by GitHub
commit bfe1221138
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
35 changed files with 2316 additions and 723 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/core/CMakeLists.txt
View file

@ -76,3 +76,5 @@ install(
LIBRARY DESTINATION "${CMAKE_INSTALL_LIBDIR}" COMPONENT lib
ARCHIVE DESTINATION "${CMAKE_INSTALL_LIBDIR}" COMPONENT lib
)
add_subdirectory("${CMAKE_CURRENT_LIST_DIR}/xtests")

8
src/core/ddsc/src/dds__domain.h
View file

@ -18,11 +18,11 @@
extern "C" {
#endif
extern const ut_avlTreedef_t dds_domaintree_def;
extern DDS_EXPORT const ut_avlTreedef_t dds_domaintree_def;
extern dds_domain * dds_domain_create (dds_domainid_t id);
extern void dds_domain_free (dds_domain * domain);
extern dds_domain * dds_domain_find_locked (dds_domainid_t id);
DDS_EXPORT dds_domain * dds_domain_create (dds_domainid_t id);
DDS_EXPORT void dds_domain_free (dds_domain * domain);
DDS_EXPORT dds_domain * dds_domain_find_locked (dds_domainid_t id);
#if defined (__cplusplus)
}

43
src/core/ddsc/src/dds__entity.h
View file

@ -19,7 +19,8 @@
extern "C" {
#endif
_Check_return_ dds_entity_t
_Check_return_
DDS_EXPORT dds_entity_t
dds_entity_init(
_In_ dds_entity * e,
_When_(kind != DDS_KIND_PARTICIPANT, _Notnull_)
@ -30,10 +31,10 @@ dds_entity_init(
_In_opt_ const dds_listener_t *listener,
_In_ uint32_t mask);
void
DDS_EXPORT void
dds_entity_add_ref(
_In_ dds_entity *e);
void
DDS_EXPORT void
dds_entity_add_ref_nolock(
_In_ dds_entity *e);
@ -56,82 +57,82 @@ dds_entity_add_ref_nolock(
qualifier_ dds__retcode_t type_##_lock (dds_entity_t hdl, type_ **x); \
qualifier_ void type_##_unlock (type_ *x);
inline bool dds_entity_is_enabled (const dds_entity *e) {
DDS_EXPORT inline bool dds_entity_is_enabled (const dds_entity *e) {
return (e->m_flags & DDS_ENTITY_ENABLED) != 0;
}
void dds_entity_status_set (dds_entity *e, uint32_t t);
DDS_EXPORT void dds_entity_status_set (dds_entity *e, uint32_t t);
inline void dds_entity_status_reset (dds_entity *e, uint32_t t) {
DDS_EXPORT inline void dds_entity_status_reset (dds_entity *e, uint32_t t) {
e->m_trigger &= ~t;
}
inline bool dds_entity_status_match (const dds_entity *e, uint32_t t) {
DDS_EXPORT inline bool dds_entity_status_match (const dds_entity *e, uint32_t t) {
return (e->m_trigger & t) != 0;
}
inline dds_entity_kind_t dds_entity_kind (const dds_entity *e) {
DDS_EXPORT inline dds_entity_kind_t dds_entity_kind (const dds_entity *e) {
return (dds_entity_kind_t) (e->m_hdl & DDS_ENTITY_KIND_MASK);
}
inline dds_entity_kind_t dds_entity_kind_from_handle (dds_entity_t hdl) {
DDS_EXPORT inline dds_entity_kind_t dds_entity_kind_from_handle (dds_entity_t hdl) {
return (hdl > 0) ? (dds_entity_kind_t) (hdl & DDS_ENTITY_KIND_MASK) : DDS_KIND_DONTCARE;
}
void dds_entity_status_signal (dds_entity *e);
DDS_EXPORT void dds_entity_status_signal (dds_entity *e);
void dds_entity_invoke_listener (const dds_entity *entity, enum dds_status_id which, const void *vst);
DDS_EXPORT void dds_entity_invoke_listener (const dds_entity *entity, enum dds_status_id which, const void *vst);
_Check_return_ dds__retcode_t
_Check_return_ DDS_EXPORT dds__retcode_t
dds_valid_hdl(
_In_ dds_entity_t hdl,
_In_ dds_entity_kind_t kind);
_Acquires_exclusive_lock_(*e)
_Check_return_ dds__retcode_t
_Check_return_ DDS_EXPORT dds__retcode_t
dds_entity_lock(
_In_ dds_entity_t hdl,
_In_ dds_entity_kind_t kind,
_Out_ dds_entity **e);
_Releases_exclusive_lock_(e)
void
DDS_EXPORT void
dds_entity_unlock(
_Inout_ dds_entity *e);
_Check_return_ dds__retcode_t
_Check_return_ DDS_EXPORT dds__retcode_t
dds_entity_observer_register_nl(
_In_ dds_entity* observed,
_In_ dds_entity_t observer,
_In_ dds_entity_callback cb);
_Check_return_ dds__retcode_t
_Check_return_ DDS_EXPORT dds__retcode_t
dds_entity_observer_register(
_In_ dds_entity_t observed,
_In_ dds_entity_t observer,
_In_ dds_entity_callback cb);
dds__retcode_t
DDS_EXPORT dds__retcode_t
dds_entity_observer_unregister_nl(
_In_ dds_entity* observed,
_In_ dds_entity_t observer);
dds__retcode_t
DDS_EXPORT dds__retcode_t
dds_entity_observer_unregister(
_In_ dds_entity_t observed,
_In_ dds_entity_t observer);
_Pre_satisfies_(entity & DDS_ENTITY_KIND_MASK)
dds_return_t
DDS_EXPORT dds_return_t
dds_delete_impl(
_In_ dds_entity_t entity,
_In_ bool keep_if_explicit);
const char *
DDS_EXPORT const char *
dds__entity_kind_str(
_In_ dds_entity_t e);
dds_domain *
DDS_EXPORT dds_domain *
dds__entity_domain(
_In_ dds_entity* e);

3
src/core/ddsc/src/dds__readcond.h
View file

@ -18,6 +18,7 @@ _Must_inspect_result_ dds_readcond*
dds_create_readcond(
_In_ dds_reader *rd,
_In_ dds_entity_kind_t kind,
_In_ uint32_t mask);
_In_ uint32_t mask,
_In_opt_ dds_querycondition_filter_fn filter);
#endif

33
src/core/ddsc/src/dds__rhc.h
View file

@ -27,21 +27,17 @@ struct ddsi_serdata;
struct ddsi_tkmap_instance;
struct proxy_writer_info;
struct rhc * dds_rhc_new (dds_reader * reader, const struct ddsi_sertopic * topic);
void dds_rhc_free (struct rhc * rhc);
void dds_rhc_fini (struct rhc * rhc);
DDS_EXPORT struct rhc *dds_rhc_new (dds_reader *reader, const struct ddsi_sertopic *topic);
DDS_EXPORT void dds_rhc_free (struct rhc *rhc);
DDS_EXPORT void dds_rhc_fini (struct rhc *rhc);
uint32_t dds_rhc_lock_samples (struct rhc * rhc);
DDS_EXPORT uint32_t dds_rhc_lock_samples (struct rhc *rhc);
DDS_EXPORT bool dds_rhc_store
(
struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info,
struct ddsi_serdata * __restrict sample, struct ddsi_tkmap_instance * __restrict tk
);
void dds_rhc_unregister_wr (struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info);
void dds_rhc_relinquish_ownership (struct rhc * __restrict rhc, const uint64_t wr_iid);
DDS_EXPORT bool dds_rhc_store (struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info, struct ddsi_serdata * __restrict sample, struct ddsi_tkmap_instance * __restrict tk);
DDS_EXPORT void dds_rhc_unregister_wr (struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info);
DDS_EXPORT void dds_rhc_relinquish_ownership (struct rhc * __restrict rhc, const uint64_t wr_iid);
int
DDS_EXPORT int
dds_rhc_read(
struct rhc *rhc,
bool lock,
@ -51,7 +47,7 @@ dds_rhc_read(
uint32_t mask,
dds_instance_handle_t handle,
dds_readcond *cond);
int
DDS_EXPORT int
dds_rhc_take(
struct rhc *rhc,
bool lock,
@ -62,15 +58,12 @@ dds_rhc_take(
dds_instance_handle_t handle,
dds_readcond *cond);
void dds_rhc_set_qos (struct rhc * rhc, const struct nn_xqos * qos);
DDS_EXPORT void dds_rhc_set_qos (struct rhc * rhc, const struct nn_xqos * qos);
void dds_rhc_add_readcondition (dds_readcond * cond);
void dds_rhc_remove_readcondition (dds_readcond * cond);
DDS_EXPORT bool dds_rhc_add_readcondition (dds_readcond * cond);
DDS_EXPORT void dds_rhc_remove_readcondition (dds_readcond * cond);
bool dds_rhc_add_waitset (dds_readcond * cond, dds_waitset * waitset, dds_attach_t x);
int dds_rhc_remove_waitset (dds_readcond * cond, dds_waitset * waitset);
int dds_rhc_takecdr
DDS_EXPORT int dds_rhc_takecdr
(
struct rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq,
uint32_t max_samples, unsigned sample_states,

4
src/core/ddsc/src/dds__topic.h
View file

@ -21,8 +21,8 @@ extern "C" {
DEFINE_ENTITY_LOCK_UNLOCK(inline, dds_topic, DDS_KIND_TOPIC)
extern struct ddsi_sertopic * dds_topic_lookup (dds_domain * domain, const char * name);
extern void dds_topic_free (dds_domainid_t domainid, struct ddsi_sertopic * st);
DDS_EXPORT struct ddsi_sertopic * dds_topic_lookup (dds_domain * domain, const char * name);
DDS_EXPORT void dds_topic_free (dds_domainid_t domainid, struct ddsi_sertopic * st);
#ifndef DDS_TOPIC_INTERN_FILTER_FN_DEFINED
#define DDS_TOPIC_INTERN_FILTER_FN_DEFINED

5
src/core/ddsc/src/dds__types.h
View file

@ -234,6 +234,8 @@ typedef struct dds_topic
}
dds_topic;
typedef uint32_t dds_querycond_mask_t;
typedef struct dds_readcond
{
dds_entity m_entity;
@ -243,10 +245,11 @@ typedef struct dds_readcond
uint32_t m_view_states;
uint32_t m_instance_states;
nn_guid_t m_rd_guid;
struct dds_readcond * m_rhc_next;
struct dds_readcond * m_next;
struct
{
dds_querycondition_filter_fn m_filter;
dds_querycond_mask_t m_qcmask; /* condition mask in RHC*/
} m_query;
}
dds_readcond;

2
src/core/ddsc/src/dds_entity.c
View file

@ -164,7 +164,7 @@ dds_return_t dds_delete_impl (dds_entity_t entity, bool keep_if_explicit)
rc = dds_entity_lock (entity, UT_HANDLE_DONTCARE_KIND, &e);
if (rc != DDS_RETCODE_OK)
{
DDS_ERROR ("Error on locking entity\n");
DDS_TRACE ("dds_delete_impl: error on locking entity %"PRIu32" keep_if_explicit %d\n", entity, (int) keep_if_explicit);
return DDS_ERRNO (rc);
}

11
src/core/ddsc/src/dds_querycond.c
View file

@ -32,11 +32,16 @@ dds_create_querycondition(
rc = dds_reader_lock(reader, &r);
if (rc == DDS_RETCODE_OK) {
dds_readcond *cond = dds_create_readcond(r, DDS_KIND_COND_QUERY, mask);
dds_readcond *cond = dds_create_readcond(r, DDS_KIND_COND_QUERY, mask, filter);
assert(cond);
hdl = cond->m_entity.m_hdl;
cond->m_query.m_filter = filter;
const bool success = (cond->m_entity.m_deriver.delete != 0);
dds_reader_unlock(r);
if (success) {
hdl = cond->m_entity.m_hdl;
} else {
dds_delete (cond->m_entity.m_hdl);
hdl = DDS_ERRNO(DDS_RETCODE_OUT_OF_RESOURCES);
}
} else {
DDS_ERROR("Error occurred on locking reader\n");
hdl = DDS_ERRNO(rc);

19
src/core/ddsc/src/dds_readcond.c
View file

@ -31,10 +31,11 @@ _Must_inspect_result_ dds_readcond*
dds_create_readcond(
_In_ dds_reader *rd,
_In_ dds_entity_kind_t kind,
_In_ uint32_t mask)
_In_ uint32_t mask,
_In_opt_ dds_querycondition_filter_fn filter)
{
dds_readcond * cond = dds_alloc(sizeof(*cond));
assert(kind == DDS_KIND_COND_READ || kind == DDS_KIND_COND_QUERY);
assert((kind == DDS_KIND_COND_READ && filter == 0) || (kind == DDS_KIND_COND_QUERY && filter != 0));
cond->m_entity.m_hdl = dds_entity_init(&cond->m_entity, (dds_entity*)rd, kind, NULL, NULL, 0);
cond->m_entity.m_deriver.delete = dds_readcond_delete;
cond->m_rhc = rd->m_rd->rhc;
@ -42,7 +43,16 @@ dds_create_readcond(
cond->m_view_states = mask & DDS_ANY_VIEW_STATE;
cond->m_instance_states = mask & DDS_ANY_INSTANCE_STATE;
cond->m_rd_guid = rd->m_entity.m_guid;
dds_rhc_add_readcondition (cond);
if (kind == DDS_KIND_COND_QUERY) {
cond->m_query.m_filter = filter;
cond->m_query.m_qcmask = 0;
}
if (!dds_rhc_add_readcondition (cond)) {
/* FIXME: current entity management code can't deal with an error late in the creation of the
entity because it doesn't allow deleting it again ... instead use a hack to signal a problem
to the caller and let that one handle it. */
cond->m_entity.m_deriver.delete = 0;
}
return cond;
}
@ -58,8 +68,9 @@ dds_create_readcondition(
rc = dds_reader_lock(reader, &rd);
if (rc == DDS_RETCODE_OK) {
dds_readcond *cond = dds_create_readcond(rd, DDS_KIND_COND_READ, mask);
dds_readcond *cond = dds_create_readcond(rd, DDS_KIND_COND_READ, mask, 0);
assert(cond);
assert(cond->m_entity.m_deriver.delete);
hdl = cond->m_entity.m_hdl;
dds_reader_unlock(rd);
} else {

1323
src/core/ddsc/src/dds_rhc.c
View file

File diff suppressed because it is too large Load diff

2
src/core/ddsc/src/dds_stream.c
View file

@ -303,7 +303,7 @@ char * dds_stream_reuse_string
}
else
{
if ((str == NULL) || (strlen (str) < length))
if ((str == NULL) || (strlen (str) + 1 < length))
{
str = dds_realloc (str, length);
}

5
src/core/ddsc/tests/CMakeLists.txt
View file

@ -50,8 +50,9 @@ set(ddsc_test_sources
"writer.c")
add_cunit_executable(cunit_ddsc ${ddsc_test_sources})
target_include_directories(cunit_ddsc PRIVATE
"$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}/src/include/>")
target_include_directories(
cunit_ddsc PRIVATE
"$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}/src/include/>")
target_link_libraries(cunit_ddsc RoundTrip Space TypesArrayKey ddsc OSAPI)
# Setup environment for config-tests

7
src/core/ddsi/include/ddsi/ddsi_iid.h
View file

@ -13,6 +13,7 @@
#define _DDS_IID_H_
#include "os/os.h"
#include "ddsc/dds_export.h"
#if defined (__cplusplus)
extern "C" {
@ -28,9 +29,9 @@ struct ddsi_iid {
uint32_t key[4];
};
void ddsi_iid_init (void);
void ddsi_iid_fini (void);
uint64_t ddsi_iid_gen (void);
DDS_EXPORT void ddsi_iid_init (void);
DDS_EXPORT void ddsi_iid_fini (void);
DDS_EXPORT uint64_t ddsi_iid_gen (void);
#if defined (__cplusplus)
}

2
src/core/ddsi/include/ddsi/ddsi_rhc_plugin.h
View file

@ -41,6 +41,6 @@ struct ddsi_rhc_plugin
void (*rhc_set_qos_fn) (struct rhc * rhc, const struct nn_xqos * qos);
};
void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct nn_xqos *xqos);
DDS_EXPORT void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct nn_xqos *xqos);
#endif

28
src/core/ddsi/include/ddsi/ddsi_serdata.h
View file

@ -108,60 +108,60 @@ struct ddsi_serdata_ops {
ddsi_serdata_free_t free;
};
void ddsi_serdata_init (struct ddsi_serdata *d, const struct ddsi_sertopic *tp, enum ddsi_serdata_kind kind);
DDS_EXPORT void ddsi_serdata_init (struct ddsi_serdata *d, const struct ddsi_sertopic *tp, enum ddsi_serdata_kind kind);
inline struct ddsi_serdata *ddsi_serdata_ref (const struct ddsi_serdata *serdata_const) {
DDS_EXPORT inline struct ddsi_serdata *ddsi_serdata_ref (const struct ddsi_serdata *serdata_const) {
struct ddsi_serdata *serdata = (struct ddsi_serdata *)serdata_const;
os_atomic_inc32 (&serdata->refc);
return serdata;
}
inline void ddsi_serdata_unref (struct ddsi_serdata *serdata) {
DDS_EXPORT inline void ddsi_serdata_unref (struct ddsi_serdata *serdata) {
if (os_atomic_dec32_ov (&serdata->refc) == 1)
serdata->ops->free (serdata);
}
inline uint32_t ddsi_serdata_size (const struct ddsi_serdata *d) {
DDS_EXPORT inline uint32_t ddsi_serdata_size (const struct ddsi_serdata *d) {
return d->ops->get_size (d);
}
inline struct ddsi_serdata *ddsi_serdata_from_ser (const struct ddsi_sertopic *topic, enum ddsi_serdata_kind kind, const struct nn_rdata *fragchain, size_t size) {
DDS_EXPORT inline struct ddsi_serdata *ddsi_serdata_from_ser (const struct ddsi_sertopic *topic, enum ddsi_serdata_kind kind, const struct nn_rdata *fragchain, size_t size) {
return topic->serdata_ops->from_ser (topic, kind, fragchain, size);
}
inline struct ddsi_serdata *ddsi_serdata_from_keyhash (const struct ddsi_sertopic *topic, const struct nn_keyhash *keyhash) {
DDS_EXPORT inline struct ddsi_serdata *ddsi_serdata_from_keyhash (const struct ddsi_sertopic *topic, const struct nn_keyhash *keyhash) {
return topic->serdata_ops->from_keyhash (topic, keyhash);
}
inline struct ddsi_serdata *ddsi_serdata_from_sample (const struct ddsi_sertopic *topic, enum ddsi_serdata_kind kind, const void *sample) {
DDS_EXPORT inline struct ddsi_serdata *ddsi_serdata_from_sample (const struct ddsi_sertopic *topic, enum ddsi_serdata_kind kind, const void *sample) {
return topic->serdata_ops->from_sample (topic, kind, sample);
}
inline struct ddsi_serdata *ddsi_serdata_to_topicless (const struct ddsi_serdata *d) {
DDS_EXPORT inline struct ddsi_serdata *ddsi_serdata_to_topicless (const struct ddsi_serdata *d) {
return d->ops->to_topicless (d);
}
inline void ddsi_serdata_to_ser (const struct ddsi_serdata *d, size_t off, size_t sz, void *buf) {
DDS_EXPORT inline void ddsi_serdata_to_ser (const struct ddsi_serdata *d, size_t off, size_t sz, void *buf) {
d->ops->to_ser (d, off, sz, buf);
}
inline struct ddsi_serdata *ddsi_serdata_to_ser_ref (const struct ddsi_serdata *d, size_t off, size_t sz, os_iovec_t *ref) {
DDS_EXPORT inline struct ddsi_serdata *ddsi_serdata_to_ser_ref (const struct ddsi_serdata *d, size_t off, size_t sz, os_iovec_t *ref) {
return d->ops->to_ser_ref (d, off, sz, ref);
}
inline void ddsi_serdata_to_ser_unref (struct ddsi_serdata *d, const os_iovec_t *ref) {
DDS_EXPORT inline void ddsi_serdata_to_ser_unref (struct ddsi_serdata *d, const os_iovec_t *ref) {
d->ops->to_ser_unref (d, ref);
}
inline bool ddsi_serdata_to_sample (const struct ddsi_serdata *d, void *sample, void **bufptr, void *buflim) {
DDS_EXPORT inline bool ddsi_serdata_to_sample (const struct ddsi_serdata *d, void *sample, void **bufptr, void *buflim) {
return d->ops->to_sample (d, sample, bufptr, buflim);
}
inline bool ddsi_serdata_topicless_to_sample (const struct ddsi_sertopic *topic, const struct ddsi_serdata *d, void *sample, void **bufptr, void *buflim) {
DDS_EXPORT inline bool ddsi_serdata_topicless_to_sample (const struct ddsi_sertopic *topic, const struct ddsi_serdata *d, void *sample, void **bufptr, void *buflim) {
return d->ops->topicless_to_sample (topic, d, sample, bufptr, buflim);
}
inline bool ddsi_serdata_eqkey (const struct ddsi_serdata *a, const struct ddsi_serdata *b) {
DDS_EXPORT inline bool ddsi_serdata_eqkey (const struct ddsi_serdata *a, const struct ddsi_serdata *b) {
return a->ops->eqkey (a, b);
}

10
src/core/ddsi/include/ddsi/ddsi_serdata_default.h
View file

@ -125,12 +125,12 @@ struct ddsi_rawcdr_sample {
size_t keysize;
};
extern const struct ddsi_sertopic_ops ddsi_sertopic_ops_default;
extern DDS_EXPORT const struct ddsi_sertopic_ops ddsi_sertopic_ops_default;
extern const struct ddsi_serdata_ops ddsi_serdata_ops_cdr;
extern const struct ddsi_serdata_ops ddsi_serdata_ops_cdr_nokey;
extern const struct ddsi_serdata_ops ddsi_serdata_ops_plist;
extern const struct ddsi_serdata_ops ddsi_serdata_ops_rawcdr;
extern DDS_EXPORT const struct ddsi_serdata_ops ddsi_serdata_ops_cdr;
extern DDS_EXPORT const struct ddsi_serdata_ops ddsi_serdata_ops_cdr_nokey;
extern DDS_EXPORT const struct ddsi_serdata_ops ddsi_serdata_ops_plist;
extern DDS_EXPORT const struct ddsi_serdata_ops ddsi_serdata_ops_rawcdr;
struct serdatapool * ddsi_serdatapool_new (void);
void ddsi_serdatapool_free (struct serdatapool * pool);

20
src/core/ddsi/include/ddsi/ddsi_sertopic.h
View file

@ -56,32 +56,32 @@ struct ddsi_sertopic_ops {
ddsi_sertopic_free_samples_t free_samples;
};
struct ddsi_sertopic *ddsi_sertopic_ref (const struct ddsi_sertopic *tp);
void ddsi_sertopic_unref (struct ddsi_sertopic *tp);
uint32_t ddsi_sertopic_compute_serdata_basehash (const struct ddsi_serdata_ops *ops);
DDS_EXPORT struct ddsi_sertopic *ddsi_sertopic_ref (const struct ddsi_sertopic *tp);
DDS_EXPORT void ddsi_sertopic_unref (struct ddsi_sertopic *tp);
DDS_EXPORT uint32_t ddsi_sertopic_compute_serdata_basehash (const struct ddsi_serdata_ops *ops);
inline void ddsi_sertopic_deinit (struct ddsi_sertopic *tp) {
DDS_EXPORT inline void ddsi_sertopic_deinit (struct ddsi_sertopic *tp) {
tp->ops->deinit (tp);
}
inline void ddsi_sertopic_zero_samples (const struct ddsi_sertopic *tp, void *samples, size_t count) {
DDS_EXPORT inline void ddsi_sertopic_zero_samples (const struct ddsi_sertopic *tp, void *samples, size_t count) {
tp->ops->zero_samples (tp, samples, count);
}
inline void ddsi_sertopic_realloc_samples (void **ptrs, const struct ddsi_sertopic *tp, void *old, size_t oldcount, size_t count)
DDS_EXPORT inline void ddsi_sertopic_realloc_samples (void **ptrs, const struct ddsi_sertopic *tp, void *old, size_t oldcount, size_t count)
{
tp->ops->realloc_samples (ptrs, tp, old, oldcount, count);
}
inline void ddsi_sertopic_free_samples (const struct ddsi_sertopic *tp, void **ptrs, size_t count, dds_free_op_t op) {
DDS_EXPORT inline void ddsi_sertopic_free_samples (const struct ddsi_sertopic *tp, void **ptrs, size_t count, dds_free_op_t op) {
tp->ops->free_samples (tp, ptrs, count, op);
}
inline void ddsi_sertopic_zero_sample (const struct ddsi_sertopic *tp, void *sample) {
DDS_EXPORT inline void ddsi_sertopic_zero_sample (const struct ddsi_sertopic *tp, void *sample) {
ddsi_sertopic_zero_samples (tp, sample, 1);
}
inline void *ddsi_sertopic_alloc_sample (const struct ddsi_sertopic *tp) {
DDS_EXPORT inline void *ddsi_sertopic_alloc_sample (const struct ddsi_sertopic *tp) {
void *ptr;
ddsi_sertopic_realloc_samples (&ptr, tp, NULL, 0, 1);
return ptr;
}
inline void ddsi_sertopic_free_sample (const struct ddsi_sertopic *tp, void *sample, dds_free_op_t op) {
DDS_EXPORT inline void ddsi_sertopic_free_sample (const struct ddsi_sertopic *tp, void *sample, dds_free_op_t op) {
ddsi_sertopic_free_samples (tp, &sample, 1, op);
}

21
src/core/ddsi/include/ddsi/ddsi_tkmap.h
View file

@ -30,19 +30,14 @@ struct ddsi_tkmap_instance
os_atomic_uint32_t m_refc;
};
struct ddsi_tkmap * ddsi_tkmap_new (void);
void ddsi_tkmap_free (_Inout_ _Post_invalid_ struct ddsi_tkmap *tkmap);
void ddsi_tkmap_instance_ref (_In_ struct ddsi_tkmap_instance *tk);
uint64_t ddsi_tkmap_lookup (_In_ struct ddsi_tkmap *tkmap, _In_ const struct ddsi_serdata *serdata);
_Check_return_ struct ddsi_tkmap_instance * ddsi_tkmap_find(
_In_ struct ddsi_serdata * sd,
_In_ const bool rd,
_In_ const bool create);
_Check_return_ struct ddsi_tkmap_instance * ddsi_tkmap_find_by_id (_In_ struct ddsi_tkmap * map, _In_ uint64_t iid);
DDS_EXPORT _Check_return_ struct ddsi_tkmap_instance * ddsi_tkmap_lookup_instance_ref (_In_ struct ddsi_serdata * sd);
DDS_EXPORT void ddsi_tkmap_instance_unref (_In_ struct ddsi_tkmap_instance * tk);
DDS_EXPORT struct ddsi_tkmap * ddsi_tkmap_new (void);
DDS_EXPORT void ddsi_tkmap_free (_Inout_ _Post_invalid_ struct ddsi_tkmap *tkmap);
DDS_EXPORT void ddsi_tkmap_instance_ref (_In_ struct ddsi_tkmap_instance *tk);
DDS_EXPORT uint64_t ddsi_tkmap_lookup (_In_ struct ddsi_tkmap *tkmap, _In_ const struct ddsi_serdata *serdata);
DDS_EXPORT struct ddsi_tkmap_instance * ddsi_tkmap_find(struct ddsi_serdata *sd, const bool rd, const bool create);
DDS_EXPORT struct ddsi_tkmap_instance * ddsi_tkmap_find_by_id (struct ddsi_tkmap *map, uint64_t iid);
DDS_EXPORT struct ddsi_tkmap_instance * ddsi_tkmap_lookup_instance_ref (struct ddsi_serdata * sd);
DDS_EXPORT void ddsi_tkmap_instance_unref (struct ddsi_tkmap_instance *tk);
#if defined (__cplusplus)
}

18
src/core/ddsi/include/ddsi/ddsi_vendor.h
View file

@ -52,26 +52,26 @@ inline bool vendor_is_eclipse (nn_vendorid_t vendor) {
return vendor_equals (vendor, NN_VENDORID_ECLIPSE);
}
inline bool vendor_is_rti (nn_vendorid_t vendor) {
return vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_RTI });
return vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_RTI }});
}
inline bool vendor_is_opensplice (nn_vendorid_t vendor) {
return vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_PRISMTECH_OSPL });
return vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_PRISMTECH_OSPL }});
}
inline bool vendor_is_twinoaks (nn_vendorid_t vendor) {
return vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_TWINOAKS });
return vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_TWINOAKS }});
}
inline bool vendor_is_cloud (nn_vendorid_t vendor) {
return vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_PRISMTECH_CLOUD });
return vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_PRISMTECH_CLOUD }});
}
inline bool vendor_is_eclipse_or_opensplice (nn_vendorid_t vendor) {
return vendor_is_eclipse (vendor) | vendor_is_opensplice (vendor);
}
inline bool vendor_is_prismtech (nn_vendorid_t vendor) {
return (vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_PRISMTECH_OSPL }) ||
vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_PRISMTECH_LITE }) ||
vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_PRISMTECH_GATEWAY }) ||
vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_PRISMTECH_JAVA }) ||
vendor_equals (vendor, (nn_vendorid_t) { 0x01, NN_VENDORID_MINOR_PRISMTECH_CLOUD }));
return (vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_PRISMTECH_OSPL }}) ||
vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_PRISMTECH_LITE }}) ||
vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_PRISMTECH_GATEWAY }}) ||
vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_PRISMTECH_JAVA }}) ||
vendor_equals (vendor, (nn_vendorid_t) {{ 0x01, NN_VENDORID_MINOR_PRISMTECH_CLOUD }}));
}
inline bool vendor_is_eclipse_or_prismtech (nn_vendorid_t vendor) {
return vendor_is_eclipse (vendor) || vendor_is_prismtech (vendor);

15
src/core/ddsi/include/ddsi/q_gc.h
View file

@ -12,6 +12,7 @@
#ifndef Q_GC_H
#define Q_GC_H
#include "ddsc/dds_export.h"
#include "ddsi/q_thread.h"
#if defined (__cplusplus)
@ -42,14 +43,14 @@ struct gcreq {
struct idx_vtime vtimes[1 /* really a flex ary */];
};
struct gcreq_queue *gcreq_queue_new (void);
void gcreq_queue_drain (struct gcreq_queue *q);
void gcreq_queue_free (struct gcreq_queue *q);
DDS_EXPORT struct gcreq_queue *gcreq_queue_new (void);
DDS_EXPORT void gcreq_queue_drain (struct gcreq_queue *q);
DDS_EXPORT void gcreq_queue_free (struct gcreq_queue *q);
struct gcreq *gcreq_new (struct gcreq_queue *gcreq_queue, gcreq_cb_t cb);
void gcreq_free (struct gcreq *gcreq);
void gcreq_enqueue (struct gcreq *gcreq);
int gcreq_requeue (struct gcreq *gcreq, gcreq_cb_t cb);
DDS_EXPORT struct gcreq *gcreq_new (struct gcreq_queue *gcreq_queue, gcreq_cb_t cb);
DDS_EXPORT void gcreq_free (struct gcreq *gcreq);
DDS_EXPORT void gcreq_enqueue (struct gcreq *gcreq);
DDS_EXPORT int gcreq_requeue (struct gcreq *gcreq, gcreq_cb_t cb);
#if defined (__cplusplus)
}

4
src/core/ddsi/include/ddsi/q_globals.h
View file

@ -18,6 +18,8 @@
#include "util/ut_fibheap.h"
#include "ddsc/dds_export.h"
#include "ddsi/q_plist.h"
#include "ddsi/q_protocol.h"
#include "ddsi/q_nwif.h"
@ -311,7 +313,7 @@ struct q_globals {
struct nn_group_membership *mship;
};
extern struct q_globals OSAPI_EXPORT gv;
extern struct q_globals DDS_EXPORT gv;
#if defined (__cplusplus)
}

46
src/core/ddsi/include/ddsi/q_plist.h
View file

@ -204,36 +204,36 @@ typedef struct nn_plist_src {
size_t bufsz;
} nn_plist_src_t;
void nn_plist_init_empty (nn_plist_t *dest);
void nn_plist_mergein_missing (nn_plist_t *a, const nn_plist_t *b);
void nn_plist_copy (nn_plist_t *dst, const nn_plist_t *src);
nn_plist_t *nn_plist_dup (const nn_plist_t *src);
int nn_plist_init_frommsg (nn_plist_t *dest, char **nextafterplist, uint64_t pwanted, uint64_t qwanted, const nn_plist_src_t *src);
void nn_plist_fini (nn_plist_t *ps);
void nn_plist_addtomsg (struct nn_xmsg *m, const nn_plist_t *ps, uint64_t pwanted, uint64_t qwanted);
int nn_plist_init_default_participant (nn_plist_t *plist);
DDS_EXPORT void nn_plist_init_empty (nn_plist_t *dest);
DDS_EXPORT void nn_plist_mergein_missing (nn_plist_t *a, const nn_plist_t *b);
DDS_EXPORT void nn_plist_copy (nn_plist_t *dst, const nn_plist_t *src);
DDS_EXPORT nn_plist_t *nn_plist_dup (const nn_plist_t *src);
DDS_EXPORT int nn_plist_init_frommsg (nn_plist_t *dest, char **nextafterplist, uint64_t pwanted, uint64_t qwanted, const nn_plist_src_t *src);
DDS_EXPORT void nn_plist_fini (nn_plist_t *ps);
DDS_EXPORT void nn_plist_addtomsg (struct nn_xmsg *m, const nn_plist_t *ps, uint64_t pwanted, uint64_t qwanted);
DDS_EXPORT int nn_plist_init_default_participant (nn_plist_t *plist);
int validate_history_qospolicy (const nn_history_qospolicy_t *q);
int validate_durability_qospolicy (const nn_durability_qospolicy_t *q);
int validate_resource_limits_qospolicy (const nn_resource_limits_qospolicy_t *q);
int validate_history_and_resource_limits (const nn_history_qospolicy_t *qh, const nn_resource_limits_qospolicy_t *qr);
int validate_durability_service_qospolicy (const nn_durability_service_qospolicy_t *q);
int validate_liveliness_qospolicy (const nn_liveliness_qospolicy_t *q);
int validate_destination_order_qospolicy (const nn_destination_order_qospolicy_t *q);
int validate_ownership_qospolicy (const nn_ownership_qospolicy_t *q);
int validate_ownership_strength_qospolicy (const nn_ownership_strength_qospolicy_t *q);
int validate_presentation_qospolicy (const nn_presentation_qospolicy_t *q);
int validate_transport_priority_qospolicy (const nn_transport_priority_qospolicy_t *q);
int validate_reader_data_lifecycle (const nn_reader_data_lifecycle_qospolicy_t *q);
int validate_duration (const nn_duration_t *d);
DDS_EXPORT int validate_history_qospolicy (const nn_history_qospolicy_t *q);
DDS_EXPORT int validate_durability_qospolicy (const nn_durability_qospolicy_t *q);
DDS_EXPORT int validate_resource_limits_qospolicy (const nn_resource_limits_qospolicy_t *q);
DDS_EXPORT int validate_history_and_resource_limits (const nn_history_qospolicy_t *qh, const nn_resource_limits_qospolicy_t *qr);
DDS_EXPORT int validate_durability_service_qospolicy (const nn_durability_service_qospolicy_t *q);
DDS_EXPORT int validate_liveliness_qospolicy (const nn_liveliness_qospolicy_t *q);
DDS_EXPORT int validate_destination_order_qospolicy (const nn_destination_order_qospolicy_t *q);
DDS_EXPORT int validate_ownership_qospolicy (const nn_ownership_qospolicy_t *q);
DDS_EXPORT int validate_ownership_strength_qospolicy (const nn_ownership_strength_qospolicy_t *q);
DDS_EXPORT int validate_presentation_qospolicy (const nn_presentation_qospolicy_t *q);
DDS_EXPORT int validate_transport_priority_qospolicy (const nn_transport_priority_qospolicy_t *q);
DDS_EXPORT int validate_reader_data_lifecycle (const nn_reader_data_lifecycle_qospolicy_t *q);
DDS_EXPORT int validate_duration (const nn_duration_t *d);
struct nn_rmsg;
struct nn_rsample_info;
struct nn_rdata;
unsigned char *nn_plist_quickscan (struct nn_rsample_info *dest, const struct nn_rmsg *rmsg, const nn_plist_src_t *src);
const unsigned char *nn_plist_findparam_native_unchecked (const void *src, nn_parameterid_t pid);
DDS_EXPORT unsigned char *nn_plist_quickscan (struct nn_rsample_info *dest, const struct nn_rmsg *rmsg, const nn_plist_src_t *src);
DDS_EXPORT const unsigned char *nn_plist_findparam_native_unchecked (const void *src, nn_parameterid_t pid);
#if defined (__cplusplus)
}

50
src/core/ddsi/include/ddsi/q_thread.h
View file

@ -13,6 +13,7 @@
#define Q_THREAD_H
#include "os/os.h"
#include "ddsc/dds_export.h"
#include "ddsi/q_static_assert.h"
#if defined (__cplusplus)
@ -81,43 +82,43 @@ struct thread_states {
struct thread_state1 *ts; /* [nthreads] */
};
extern struct thread_states thread_states;
extern DDS_EXPORT struct thread_states thread_states;
extern os_threadLocal struct thread_state1 *tsd_thread_state;
void thread_states_init_static (void);
void thread_states_init (_In_ unsigned maxthreads);
void thread_states_fini (void);
DDS_EXPORT void thread_states_init_static (void);
DDS_EXPORT void thread_states_init (_In_ unsigned maxthreads);
DDS_EXPORT void thread_states_fini (void);
void upgrade_main_thread (void);
void downgrade_main_thread (void);
const struct config_thread_properties_listelem *lookup_thread_properties (_In_z_ const char *name);
_Success_(return != NULL) _Ret_maybenull_ struct thread_state1 *create_thread (_In_z_ const char *name, _In_ uint32_t (*f) (void *arg), _In_opt_ void *arg);
_Ret_valid_ struct thread_state1 *lookup_thread_state (void);
_Success_(return != NULL) _Ret_maybenull_ struct thread_state1 *lookup_thread_state_real (void);
_Success_(return == 0) int join_thread (_Inout_ struct thread_state1 *ts1);
void log_stack_traces (void);
struct thread_state1 *get_thread_state (_In_ os_threadId id);
struct thread_state1 * init_thread_state (_In_z_ const char *tname);
void reset_thread_state (_Inout_opt_ struct thread_state1 *ts1);
int thread_exists (_In_z_ const char *name);
DDS_EXPORT void upgrade_main_thread (void);
DDS_EXPORT void downgrade_main_thread (void);
DDS_EXPORT const struct config_thread_properties_listelem *lookup_thread_properties (const char *name);
DDS_EXPORT struct thread_state1 *create_thread (const char *name, uint32_t (*f) (void *arg), void *arg);
DDS_EXPORT struct thread_state1 *lookup_thread_state (void);
DDS_EXPORT struct thread_state1 *lookup_thread_state_real (void);
DDS_EXPORT int join_thread (_Inout_ struct thread_state1 *ts1);
DDS_EXPORT void log_stack_traces (void);
DDS_EXPORT struct thread_state1 *get_thread_state (_In_ os_threadId id);
DDS_EXPORT struct thread_state1 * init_thread_state (_In_z_ const char *tname);
DDS_EXPORT void reset_thread_state (_Inout_opt_ struct thread_state1 *ts1);
DDS_EXPORT int thread_exists (_In_z_ const char *name);
inline int vtime_awake_p (_In_ vtime_t vtime)
DDS_EXPORT inline int vtime_awake_p (_In_ vtime_t vtime)
{
return (vtime % 2) == 0;
}
inline int vtime_asleep_p (_In_ vtime_t vtime)
DDS_EXPORT inline int vtime_asleep_p (_In_ vtime_t vtime)
{
return (vtime % 2) == 1;
}
inline int vtime_gt (_In_ vtime_t vtime1, _In_ vtime_t vtime0)
DDS_EXPORT inline int vtime_gt (_In_ vtime_t vtime1, _In_ vtime_t vtime0)
{
Q_STATIC_ASSERT_CODE (sizeof (vtime_t) == sizeof (svtime_t));
return (svtime_t) (vtime1 - vtime0) > 0;
}
inline void thread_state_asleep (_Inout_ struct thread_state1 *ts1)
DDS_EXPORT inline void thread_state_asleep (_Inout_ struct thread_state1 *ts1)
{
vtime_t vt = ts1->vtime;
vtime_t wd = ts1->watchdog;
@ -138,9 +139,9 @@ inline void thread_state_asleep (_Inout_ struct thread_state1 *ts1)
} else {
ts1->watchdog = wd + 1;
}
}
}
inline void thread_state_awake (_Inout_ struct thread_state1 *ts1)
DDS_EXPORT inline void thread_state_awake (_Inout_ struct thread_state1 *ts1)
{
vtime_t vt = ts1->vtime;
vtime_t wd = ts1->watchdog;
@ -158,10 +159,9 @@ inline void thread_state_awake (_Inout_ struct thread_state1 *ts1)
} else {
ts1->watchdog = wd + 2;
}
}
inline void thread_state_blocked (_Inout_ struct thread_state1 *ts1)
DDS_EXPORT inline void thread_state_blocked (_Inout_ struct thread_state1 *ts1)
{
vtime_t wd = ts1->watchdog;
if ( wd % 2 ){
@ -171,7 +171,7 @@ inline void thread_state_blocked (_Inout_ struct thread_state1 *ts1)
}
}
inline void thread_state_unblocked (_Inout_ struct thread_state1 *ts1)
DDS_EXPORT inline void thread_state_unblocked (_Inout_ struct thread_state1 *ts1)
{
vtime_t wd = ts1->watchdog;
if ( wd % 2 ){

29
src/core/ddsi/include/ddsi/q_time.h
View file

@ -13,6 +13,7 @@
#define NN_TIME_H
#include "os/os.h"
#include "ddsc/dds_export.h"
#if defined (__cplusplus)
extern "C" {
@ -55,21 +56,21 @@ extern const nn_duration_t duration_infinite;
int valid_ddsi_timestamp (nn_ddsi_time_t t);
OSAPI_EXPORT nn_wctime_t now (void); /* wall clock time */
nn_mtime_t now_mt (void); /* monotonic time */
nn_etime_t now_et (void); /* elapsed time */
void mtime_to_sec_usec (_Out_ int * __restrict sec, _Out_ int * __restrict usec, _In_ nn_mtime_t t);
void wctime_to_sec_usec (_Out_ int * __restrict sec, _Out_ int * __restrict usec, _In_ nn_wctime_t t);
void etime_to_sec_usec (_Out_ int * __restrict sec, _Out_ int * __restrict usec, _In_ nn_etime_t t);
nn_mtime_t mtime_round_up (nn_mtime_t t, int64_t round);
nn_mtime_t add_duration_to_mtime (nn_mtime_t t, int64_t d);
nn_wctime_t add_duration_to_wctime (nn_wctime_t t, int64_t d);
nn_etime_t add_duration_to_etime (nn_etime_t t, int64_t d);
DDS_EXPORT nn_wctime_t now (void); /* wall clock time */
DDS_EXPORT nn_mtime_t now_mt (void); /* monotonic time */
DDS_EXPORT nn_etime_t now_et (void); /* elapsed time */
DDS_EXPORT void mtime_to_sec_usec (_Out_ int * __restrict sec, _Out_ int * __restrict usec, _In_ nn_mtime_t t);
DDS_EXPORT void wctime_to_sec_usec (_Out_ int * __restrict sec, _Out_ int * __restrict usec, _In_ nn_wctime_t t);
DDS_EXPORT void etime_to_sec_usec (_Out_ int * __restrict sec, _Out_ int * __restrict usec, _In_ nn_etime_t t);
DDS_EXPORT nn_mtime_t mtime_round_up (nn_mtime_t t, int64_t round);
DDS_EXPORT nn_mtime_t add_duration_to_mtime (nn_mtime_t t, int64_t d);
DDS_EXPORT nn_wctime_t add_duration_to_wctime (nn_wctime_t t, int64_t d);
DDS_EXPORT nn_etime_t add_duration_to_etime (nn_etime_t t, int64_t d);
nn_ddsi_time_t nn_wctime_to_ddsi_time (nn_wctime_t t);
OSAPI_EXPORT nn_wctime_t nn_wctime_from_ddsi_time (nn_ddsi_time_t x);
OSAPI_EXPORT nn_duration_t nn_to_ddsi_duration (int64_t t);
OSAPI_EXPORT int64_t nn_from_ddsi_duration (nn_duration_t x);
DDS_EXPORT nn_ddsi_time_t nn_wctime_to_ddsi_time (nn_wctime_t t);
DDS_EXPORT nn_wctime_t nn_wctime_from_ddsi_time (nn_ddsi_time_t x);
DDS_EXPORT nn_duration_t nn_to_ddsi_duration (int64_t t);
DDS_EXPORT int64_t nn_from_ddsi_duration (nn_duration_t x);
#if defined (__cplusplus)
}

32
src/core/ddsi/include/ddsi/q_xevent.h
View file

@ -41,31 +41,31 @@ struct xeventq *xeventq_new
/* xeventq_free calls callback handlers with t = T_NEVER, at which point they are required to free
whatever memory is claimed for the argument and call delete_xevent. */
void xeventq_free (struct xeventq *evq);
int xeventq_start (struct xeventq *evq, const char *name); /* <0 => error, =0 => ok */
void xeventq_stop (struct xeventq *evq);
DDS_EXPORT void xeventq_free (struct xeventq *evq);
DDS_EXPORT int xeventq_start (struct xeventq *evq, const char *name); /* <0 => error, =0 => ok */
DDS_EXPORT void xeventq_stop (struct xeventq *evq);
void qxev_msg (struct xeventq *evq, struct nn_xmsg *msg);
void qxev_pwr_entityid (struct proxy_writer * pwr, nn_guid_prefix_t * id);
void qxev_prd_entityid (struct proxy_reader * prd, nn_guid_prefix_t * id);
DDS_EXPORT void qxev_msg (struct xeventq *evq, struct nn_xmsg *msg);
DDS_EXPORT void qxev_pwr_entityid (struct proxy_writer * pwr, nn_guid_prefix_t * id);
DDS_EXPORT void qxev_prd_entityid (struct proxy_reader * prd, nn_guid_prefix_t * id);
/* Returns 1 if queued, 0 otherwise (no point in returning the
event, you can't do anything with it anyway) */
int qxev_msg_rexmit_wrlock_held (struct xeventq *evq, struct nn_xmsg *msg, int force);
DDS_EXPORT int qxev_msg_rexmit_wrlock_held (struct xeventq *evq, struct nn_xmsg *msg, int force);
/* All of the following lock EVQ for the duration of the operation */
void delete_xevent (struct xevent *ev);
int resched_xevent_if_earlier (struct xevent *ev, nn_mtime_t tsched);
DDS_EXPORT void delete_xevent (struct xevent *ev);
DDS_EXPORT int resched_xevent_if_earlier (struct xevent *ev, nn_mtime_t tsched);
struct xevent *qxev_heartbeat (struct xeventq *evq, nn_mtime_t tsched, const nn_guid_t *wr_guid);
struct xevent *qxev_acknack (struct xeventq *evq, nn_mtime_t tsched, const nn_guid_t *pwr_guid, const nn_guid_t *rd_guid);
struct xevent *qxev_spdp (nn_mtime_t tsched, const nn_guid_t *pp_guid, const nn_guid_t *proxypp_guid);
struct xevent *qxev_pmd_update (nn_mtime_t tsched, const nn_guid_t *pp_guid);
struct xevent *qxev_end_startup_mode (nn_mtime_t tsched);
struct xevent *qxev_delete_writer (nn_mtime_t tsched, const nn_guid_t *guid);
DDS_EXPORT struct xevent *qxev_heartbeat (struct xeventq *evq, nn_mtime_t tsched, const nn_guid_t *wr_guid);
DDS_EXPORT struct xevent *qxev_acknack (struct xeventq *evq, nn_mtime_t tsched, const nn_guid_t *pwr_guid, const nn_guid_t *rd_guid);
DDS_EXPORT struct xevent *qxev_spdp (nn_mtime_t tsched, const nn_guid_t *pp_guid, const nn_guid_t *proxypp_guid);
DDS_EXPORT struct xevent *qxev_pmd_update (nn_mtime_t tsched, const nn_guid_t *pp_guid);
DDS_EXPORT struct xevent *qxev_end_startup_mode (nn_mtime_t tsched);
DDS_EXPORT struct xevent *qxev_delete_writer (nn_mtime_t tsched, const nn_guid_t *guid);
/* cb will be called with now = T_NEVER if the event is still enqueued when when xeventq_free starts cleaning up */
struct xevent *qxev_callback (nn_mtime_t tsched, void (*cb) (struct xevent *xev, void *arg, nn_mtime_t now), void *arg);
DDS_EXPORT struct xevent *qxev_callback (nn_mtime_t tsched, void (*cb) (struct xevent *xev, void *arg, nn_mtime_t now), void *arg);
#if defined (__cplusplus)
}

30
src/core/ddsi/include/ddsi/q_xqos.h
View file

@ -326,21 +326,21 @@ typedef struct nn_xqos {
struct nn_xmsg;
void nn_xqos_init_empty (nn_xqos_t *xqos);
void nn_xqos_init_default_reader (nn_xqos_t *xqos);
void nn_xqos_init_default_writer (nn_xqos_t *xqos);
void nn_xqos_init_default_writer_noautodispose (nn_xqos_t *xqos);
void nn_xqos_init_default_subscriber (nn_xqos_t *xqos);
void nn_xqos_init_default_publisher (nn_xqos_t *xqos);
void nn_xqos_init_default_topic (nn_xqos_t *xqos);
void nn_xqos_copy (nn_xqos_t *dst, const nn_xqos_t *src);
void nn_xqos_unalias (nn_xqos_t *xqos);
void nn_xqos_fini (nn_xqos_t *xqos);
void nn_xqos_mergein_missing (nn_xqos_t *a, const nn_xqos_t *b);
uint64_t nn_xqos_delta (const nn_xqos_t *a, const nn_xqos_t *b, uint64_t mask);
void nn_xqos_addtomsg (struct nn_xmsg *m, const nn_xqos_t *xqos, uint64_t wanted);
void nn_log_xqos (uint32_t cat, const nn_xqos_t *xqos);
nn_xqos_t *nn_xqos_dup (const nn_xqos_t *src);
DDS_EXPORT void nn_xqos_init_empty (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_init_default_reader (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_init_default_writer (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_init_default_writer_noautodispose (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_init_default_subscriber (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_init_default_publisher (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_init_default_topic (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_copy (nn_xqos_t *dst, const nn_xqos_t *src);
DDS_EXPORT void nn_xqos_unalias (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_fini (nn_xqos_t *xqos);
DDS_EXPORT void nn_xqos_mergein_missing (nn_xqos_t *a, const nn_xqos_t *b);
DDS_EXPORT uint64_t nn_xqos_delta (const nn_xqos_t *a, const nn_xqos_t *b, uint64_t mask);
DDS_EXPORT void nn_xqos_addtomsg (struct nn_xmsg *m, const nn_xqos_t *xqos, uint64_t wanted);
DDS_EXPORT void nn_log_xqos (uint32_t cat, const nn_xqos_t *xqos);
DDS_EXPORT nn_xqos_t *nn_xqos_dup (const nn_xqos_t *src);
#if defined (__cplusplus)
}

2
src/core/ddsi/src/ddsi_rhc_plugin.c
View file

@ -13,7 +13,7 @@
#include "ddsi/q_xqos.h"
#include "ddsi/ddsi_rhc_plugin.h"
void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct nn_xqos *xqos)
DDS_EXPORT void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct nn_xqos *xqos)
{
pwr_info->guid = e->guid;
pwr_info->ownership_strength = xqos->ownership_strength.value;

4
src/core/ddsi/src/q_config.c
View file

@ -97,10 +97,10 @@ struct cfgst {
/* "trace" is special: it enables (nearly) everything */
static const char *logcat_names[] = {
"fatal", "error", "warning", "info", "config", "discovery", "data", "radmin", "timing", "traffic", "topic", "tcp", "plist", "whc", "throttle", "trace", NULL
"fatal", "error", "warning", "info", "config", "discovery", "data", "radmin", "timing", "traffic", "topic", "tcp", "plist", "whc", "throttle", "rhc", "trace", NULL
};
static const uint32_t logcat_codes[] = {
DDS_LC_FATAL, DDS_LC_ERROR, DDS_LC_WARNING, DDS_LC_INFO, DDS_LC_CONFIG, DDS_LC_DISCOVERY, DDS_LC_DATA, DDS_LC_RADMIN, DDS_LC_TIMING, DDS_LC_TRAFFIC, DDS_LC_TOPIC, DDS_LC_TCP, DDS_LC_PLIST, DDS_LC_WHC, DDS_LC_THROTTLE, DDS_LC_ALL
DDS_LC_FATAL, DDS_LC_ERROR, DDS_LC_WARNING, DDS_LC_INFO, DDS_LC_CONFIG, DDS_LC_DISCOVERY, DDS_LC_DATA, DDS_LC_RADMIN, DDS_LC_TIMING, DDS_LC_TRAFFIC, DDS_LC_TOPIC, DDS_LC_TCP, DDS_LC_PLIST, DDS_LC_WHC, DDS_LC_THROTTLE, DDS_LC_RHC, DDS_LC_ALL
};
/* We want the tracing/verbosity settings to be fixed while parsing

26
src/core/xtests/CMakeLists.txt Normal file
View file

@ -0,0 +1,26 @@
#
# Copyright(c) 2019 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
#
idlc_generate(RhcTypes RhcTypes.idl)
add_executable(rhc_torture rhc_torture.c mt19937ar.c mt19937ar.h)
target_include_directories(
rhc_torture PRIVATE
"$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/../ddsc/src>"
"$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/../ddsi/include>")
target_link_libraries(rhc_torture RhcTypes ddsc util OSAPI)
add_test(
NAME rhc_torture
COMMAND rhc_torture 314159265 0 5000 0)
set_property(TEST rhc_torture PROPERTY TIMEOUT 20)

20
src/core/xtests/RhcTypes.idl Normal file
View file

@ -0,0 +1,20 @@
/*
* Copyright(c) 2019 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
*/
module RhcTypes {
struct T {
long k;
string ks;
long x, y;
string s;
};
#pragma keylist T k ks
};

184
src/core/xtests/mt19937ar.c Normal file
View file

@ -0,0 +1,184 @@
/*
* Copyright(c) 2019 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
*/
/*
A C-program for MT19937, with initialization improved 2002/1/26.
Coded by Takuji Nishimura and Makoto Matsumoto.
Before using, initialize the state by using init_genrand(seed)
or init_by_array(init_key, key_length).
Copyright (C) 1997 - 2002, Makoto Matsumoto and Takuji Nishimura,
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
3. The names of its contributors may not be used to endorse or promote
products derived from this software without specific prior written
permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
Any feedback is very welcome.
http://www.math.sci.hiroshima-u.ac.jp/~m-mat/MT/emt.html
email: m-mat @ math.sci.hiroshima-u.ac.jp (remove space)
*/
#include "mt19937ar.h"
/* Period parameters */
#define N 624
#define M 397
#define MATRIX_A 0x9908b0dfU /* constant vector a */
#define UPPER_MASK 0x80000000U /* most significant w-r bits */
#define LOWER_MASK 0x7fffffffU /* least significant r bits */
static uint32_t mt[N]; /* the array for the state vector */
static uint32_t mti=N+1; /* mti==N+1 means mt[N] is not initialized */
/* initializes mt[N] with a seed */
void init_genrand(uint32_t s)
{
mt[0]= s & 0xffffffffU;
for (mti=1; mti<N; mti++) {
mt[mti] =
(1812433253U * (mt[mti-1] ^ (mt[mti-1] >> 30)) + mti);
/* See Knuth TAOCP Vol2. 3rd Ed. P.106 for multiplier. */
/* In the previous versions, MSBs of the seed affect */
/* only MSBs of the array mt[]. */
/* 2002/01/09 modified by Makoto Matsumoto */
mt[mti] &= 0xffffffffU;
/* for >32 bit machines */
}
}
/* initialize by an array with array-length */
/* init_key is the array for initializing keys */
/* key_length is its length */
/* slight change for C++, 2004/2/26 */
void init_by_array(uint32_t init_key[], size_t key_length)
{
uint32_t i, j, k;
init_genrand(19650218U);
i=1; j=0;
k = (N>key_length ? N : (uint32_t)key_length);
for (; k; k--) {
mt[i] = (mt[i] ^ ((mt[i-1] ^ (mt[i-1] >> 30)) * 1664525U))
+ init_key[j] + j; /* non linear */
mt[i] &= 0xffffffffU; /* for WORDSIZE > 32 machines */
i++; j++;
if (i>=N) { mt[0] = mt[N-1]; i=1; }
if (j>=(uint32_t)key_length) j=0;
}
for (k=N-1; k; k--) {
mt[i] = (mt[i] ^ ((mt[i-1] ^ (mt[i-1] >> 30)) * 1566083941U))
- i; /* non linear */
mt[i] &= 0xffffffffU; /* for WORDSIZE > 32 machines */
i++;
if (i>=N) { mt[0] = mt[N-1]; i=1; }
}
mt[0] = 0x80000000U; /* MSB is 1; assuring non-zero initial array */
}
/* generates a random number on [0,0xffffffff]-interval */
uint32_t genrand_int32(void)
{
uint32_t y;
static uint32_t mag01[2]={0x0U, MATRIX_A};
/* mag01[x] = x * MATRIX_A for x=0,1 */
if (mti >= N) { /* generate N words at one time */
int kk;
if (mti == N+1) /* if init_genrand() has not been called, */
init_genrand(5489U); /* a default initial seed is used */
for (kk=0;kk<N-M;kk++) {
y = (mt[kk]&UPPER_MASK)|(mt[kk+1]&LOWER_MASK);
mt[kk] = mt[kk+M] ^ (y >> 1) ^ mag01[y & 0x1U];
}
for (;kk<N-1;kk++) {
y = (mt[kk]&UPPER_MASK)|(mt[kk+1]&LOWER_MASK);
mt[kk] = mt[kk+(M-N)] ^ (y >> 1) ^ mag01[y & 0x1U];
}
y = (mt[N-1]&UPPER_MASK)|(mt[0]&LOWER_MASK);
mt[N-1] = mt[M-1] ^ (y >> 1) ^ mag01[y & 0x1U];
mti = 0;
}
y = mt[mti++];
/* Tempering */
y ^= (y >> 11);
y ^= (y << 7) & 0x9d2c5680U;
y ^= (y << 15) & 0xefc60000U;
y ^= (y >> 18);
return (uint32_t) y;
}
/* generates a random number on [0,0x7fffffff]-interval */
int32_t genrand_int31(void)
{
return (int32_t)(genrand_int32()>>1);
}
/* generates a random number on [0,1]-real-interval */
double genrand_real1(void)
{
return genrand_int32()*(1.0/4294967295.0);
/* divided by 2^32-1 */
}
/* generates a random number on [0,1)-real-interval */
double genrand_real2(void)
{
return genrand_int32()*(1.0/4294967296.0);
/* divided by 2^32 */
}
/* generates a random number on (0,1)-real-interval */
double genrand_real3(void)
{
return (((double)genrand_int32()) + 0.5)*(1.0/4294967296.0);
/* divided by 2^32 */
}
/* generates a random number on [0,1) with 53-bit resolution*/
double genrand_res53(void)
{
uint32_t a=genrand_int32()>>5, b=genrand_int32()>>6;
return(a*67108864.0+b)*(1.0/9007199254740992.0);
}
/* These real versions are due to Isaku Wada, 2002/01/09 added */

46
src/core/xtests/mt19937ar.h Normal file
View file

@ -0,0 +1,46 @@
/*
* Copyright(c) 2019 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
*/
#ifndef MT19937AR_H
#define MT19937AR_H
#include <stdint.h>
#include <stddef.h>
/* initializes mt[N] with a seed */
void init_genrand(uint32_t s);
/* initialize by an array with array-length */
/* init_key is the array for initializing keys */
/* key_length is its length */
/* slight change for C++, 2004/2/26 */
void init_by_array(uint32_t init_key[], size_t key_length);
/* generates a random number on [0,0xffffffff]-interval */
uint32_t genrand_int32(void);
/* generates a random number on [0,0x7fffffff]-interval */
int32_t genrand_int31(void);
/* generates a random number on [0,1]-real-interval */
double genrand_real1(void);
/* generates a random number on [0,1)-real-interval */
double genrand_real2(void);
/* generates a random number on (0,1)-real-interval */
double genrand_real3(void);
/* generates a random number on [0,1) with 53-bit resolution*/
double genrand_res53(void);
/* These real versions are due to Isaku Wada, 2002/01/09 added */
#endif

983
src/core/xtests/rhc_torture.c Normal file
View file

@ -0,0 +1,983 @@
/*
* Copyright(c) 2019 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 <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <inttypes.h>
#include "os/os.h"
#include "ddsc/dds.h"
#include "ddsi/ddsi_tkmap.h"
#include "dds__entity.h"
#include "ddsi/q_config.h"
#include "ddsi/q_bswap.h"
#include "ddsi/q_globals.h"
#include "ddsi/q_radmin.h"
#include "ddsi/q_entity.h"
#include "ddsi/q_gc.h"
#include "ddsi/ddsi_serdata.h"
#include "dds__topic.h"
#include "dds__rhc.h"
#include "ddsi/ddsi_iid.h"
#include "mt19937ar.h"
#include "RhcTypes.h"
#ifndef _MSC_VER
#define STATIC_ARRAY_DIM static
#else
#define STATIC_ARRAY_DIM
#endif
static struct ddsi_sertopic *mdtopic;
static struct thread_state1 *mainthread;
static dds_time_t tref_dds;
static uint32_t seq;
static os_mutex wait_gc_cycle_lock;
static os_cond wait_gc_cycle_cond;
static int wait_gc_cycle_trig;
/* these are used to get a sufficiently large result buffer when takeing/reading everying */
#define N_KEYVALS 27
#define MAX_HIST_DEPTH 4
static dds_sample_info_t rres_iseq[(MAX_HIST_DEPTH + 1) * N_KEYVALS];
static RhcTypes_T rres_mseq[sizeof (rres_iseq) / sizeof (rres_iseq[0])];
static void *rres_ptrs[sizeof (rres_iseq) / sizeof (rres_iseq[0])];
static int64_t last_dds_time = 0;
static char *print_tstamp (char *buf, size_t sz, dds_time_t t)
{
dds_time_t d = t - tref_dds;
size_t pos = 0;
pos += (size_t) snprintf (buf + pos, sz - pos, "T");
if (d / 1000000000 != 0)
pos += (size_t) snprintf (buf + pos, sz - pos, "%+ds", (int) (d / 1000000000));
if (d % 1000000000 != 0)
snprintf (buf + pos, sz - pos, "%+dns", (int) (d % 1000000000));
return buf;
}
static int64_t dds_time_uniq (void)
{
/* behaviour depends on time stamps being strictly monotonically increasing, but there
is no way of knowing whether dds_time provides sufficient resolution, so fake it */
int64_t t = dds_time ();
if (t > last_dds_time)
last_dds_time = t;
else
last_dds_time++;
return last_dds_time;
}
static struct ddsi_serdata *mksample (int32_t keyval, unsigned statusinfo)
{
RhcTypes_T d = { keyval, "A", (int32_t) ++seq, 0, "B" };
struct ddsi_serdata *sd = ddsi_serdata_from_sample (mdtopic, SDK_DATA, &d);
sd->statusinfo = statusinfo;
sd->timestamp.v = dds_time_uniq ();
return sd;
}
static struct ddsi_serdata *mkkeysample (int32_t keyval, unsigned statusinfo)
{
RhcTypes_T d = { keyval, "A", 0, 0, "B" };
struct ddsi_serdata *sd = ddsi_serdata_from_sample (mdtopic, SDK_KEY, &d);
sd->statusinfo = statusinfo;
sd->timestamp.v = dds_time_uniq ();
return sd;
}
static uint64_t store (struct rhc *rhc, struct proxy_writer *wr, struct ddsi_serdata *sd, bool print)
{
/* beware: unrefs sd */
struct ddsi_tkmap_instance *tk;
struct proxy_writer_info pwr_info;
thread_state_awake (mainthread);
tk = ddsi_tkmap_lookup_instance_ref(sd);
uint64_t iid = tk->m_iid;
if (print)
{
RhcTypes_T d;
char buf[64];
char si_d = (sd->statusinfo & NN_STATUSINFO_DISPOSE) ? 'D' : '.';
char si_u = (sd->statusinfo & NN_STATUSINFO_UNREGISTER) ? 'U' : '.';
memset (&d, 0, sizeof (d));
ddsi_serdata_to_sample (sd, &d, NULL, NULL);
(void) print_tstamp (buf, sizeof (buf), sd->timestamp.v);
if (sd->kind == SDK_KEY)
printf ("STORE %c%c %16"PRIx64" %16"PRIx64" %2"PRId32" %6s %s\n", si_u, si_d, iid, wr->e.iid, d.k, "_", buf);
else
printf ("STORE %c%c %16"PRIx64" %16"PRIx64" %2"PRId32" %6"PRIu32" %s\n", si_u, si_d, iid, wr->e.iid, d.k, d.x, buf);
ddsi_sertopic_free_sample (sd->topic, &d, DDS_FREE_CONTENTS);
}
pwr_info.auto_dispose = wr->c.xqos->writer_data_lifecycle.autodispose_unregistered_instances;
pwr_info.guid = wr->e.guid;
pwr_info.iid = wr->e.iid;
pwr_info.ownership_strength = wr->c.xqos->ownership_strength.value;
dds_rhc_store (rhc, &pwr_info, sd, tk);
ddsi_tkmap_instance_unref (tk);
thread_state_asleep (mainthread);
ddsi_serdata_unref (sd);
return iid;
}
static struct proxy_writer *mkwr (bool auto_dispose)
{
struct proxy_writer *pwr;
struct nn_xqos *xqos;
uint64_t wr_iid;
pwr = os_malloc (sizeof (*pwr));
xqos = os_malloc (sizeof (*xqos));
wr_iid = ddsi_iid_gen ();
memset (pwr, 0, sizeof (*pwr));
nn_xqos_init_empty (xqos);
nn_xqos_mergein_missing (xqos, &gv.default_xqos_wr);
xqos->ownership_strength.value = 0;
xqos->writer_data_lifecycle.autodispose_unregistered_instances = auto_dispose;
pwr->e.iid = wr_iid;
pwr->c.xqos = xqos;
return pwr;
}
static void fwr (struct proxy_writer *wr)
{
free (wr->c.xqos);
free (wr);
}
static struct rhc *mkrhc (dds_reader *rd, nn_history_kind_t hk, int32_t hdepth, nn_destination_order_kind_t dok)
{
struct rhc *rhc;
nn_xqos_t rqos;
nn_xqos_init_empty (&rqos);
rqos.present |= QP_HISTORY | QP_DESTINATION_ORDER;
rqos.history.kind = hk;
rqos.history.depth = hdepth;
rqos.destination_order.kind = dok;
nn_xqos_mergein_missing (&rqos, &gv.default_xqos_rd);
thread_state_awake (mainthread);
rhc = dds_rhc_new (rd, mdtopic);
dds_rhc_set_qos(rhc, &rqos);
thread_state_asleep (mainthread);
return rhc;
}
static void frhc (struct rhc *rhc)
{
thread_state_awake (mainthread);
dds_rhc_free (rhc);
thread_state_asleep (mainthread);
}
static char si2is (const dds_sample_info_t *si)
{
switch (si->instance_state)
{
case DDS_ALIVE_INSTANCE_STATE: return 'A';
case DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE: return 'D';
case DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE: return 'U';
default: return '?';
}
}
static char si2ss (const dds_sample_info_t *si)
{
switch (si->sample_state)
{
case DDS_READ_SAMPLE_STATE: return 'R';
case DDS_NOT_READ_SAMPLE_STATE: return 'N';
default: return '?';
}
}
static char si2vs (const dds_sample_info_t *si)
{
switch (si->view_state)
{
case DDS_NEW_VIEW_STATE: return 'N';
case DDS_NOT_NEW_VIEW_STATE: return 'O';
default: return '?';
}
}
struct check {
const char *st;
uint64_t iid;
uint64_t wr_iid;
uint32_t dgen;
uint32_t nwgen;
int vd;
int32_t keyval;
int32_t seq;
};
static void docheck (int n, const dds_sample_info_t *iseq, const RhcTypes_T *mseq, const struct check *chk)
{
#ifndef NDEBUG
int i;
for (i = 0; i < n; i++)
{
assert (chk[i].st != 0);
dds_sample_state_t sst = chk[i].st[0] == 'N' ? DDS_NOT_READ_SAMPLE_STATE : DDS_READ_SAMPLE_STATE;
dds_view_state_t vst = chk[i].st[1] == 'O' ? DDS_NOT_NEW_VIEW_STATE : DDS_NEW_VIEW_STATE;
dds_instance_state_t ist = chk[i].st[2] == 'A' ? DDS_ALIVE_INSTANCE_STATE : chk[i].st[2] == 'U' ? DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE : DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE;
assert (iseq[i].sample_state == sst);
assert (iseq[i].view_state == vst);
assert (iseq[i].instance_state == ist);
assert (iseq[i].instance_handle == chk[i].iid);
assert (chk[i].wr_iid == 0 || iseq[i].publication_handle == chk[i].wr_iid);
assert (iseq[i].disposed_generation_count == chk[i].dgen);
assert (iseq[i].no_writers_generation_count == chk[i].nwgen);
assert (!!iseq[i].valid_data == !!chk[i].vd);
assert (mseq[i].k == chk[i].keyval);
assert (!chk[i].vd || mseq[i].x == chk[i].seq);
}
assert (chk[i].st == 0);
#else
(void)n; (void)iseq; (void)mseq; (void)chk;
#endif
}
static void print_seq (int n, const dds_sample_info_t *iseq, const RhcTypes_T *mseq)
{
int i;
printf ("INDX SVI %-16s %-16s DGEN NWRG SR GR AR KV SEQ %s\n", "INSTHANDLE", "PUBHANDLE", "TSTAMP");
for (i = 0; i < n; i++)
{
dds_sample_info_t const * const si = &iseq[i];
RhcTypes_T const * const d = &mseq[i];
char buf[64];
assert(si->instance_handle);
assert(si->publication_handle);
printf ("[%2d] %c%c%c %16"PRIx64" %16"PRIx64" %4d %4d %2d %2d %2d %2"PRId32,
i, si2ss(si), si2vs(si), si2is(si),
si->instance_handle, si->publication_handle,
si->disposed_generation_count, si->no_writers_generation_count,
si->sample_rank, si->generation_rank, si->absolute_generation_rank,
d->k);
if (si->valid_data)
printf (" %6"PRIu32, d->x);
else
printf (" %6s", "_");
printf (" %s\n", print_tstamp (buf, sizeof (buf), si->source_timestamp));
}
}
static void rdtkcond (struct rhc *rhc, dds_readcond *cond, const struct check *chk, bool print, int max, const char *opname, int (*op) (struct rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, dds_readcond *cond), uint32_t states_seen[STATIC_ARRAY_DIM 2*2*3][2])
{
int cnt;
if (print)
printf ("%s:\n", opname);
thread_state_awake (mainthread);
cnt = op (rhc, true, rres_ptrs, rres_iseq, (max <= 0) ? (uint32_t) (sizeof (rres_iseq) / sizeof (rres_iseq[0])) : (uint32_t) max, cond ? NO_STATE_MASK_SET : (DDS_ANY_SAMPLE_STATE | DDS_ANY_VIEW_STATE | DDS_ANY_INSTANCE_STATE), 0, cond);
thread_state_asleep (mainthread);
if (max > 0 && cnt > max) {
printf ("%s TOO MUCH DATA (%d > %d)\n", opname, cnt, max);
abort ();
} else if (cnt > 0) {
if (print) print_seq (cnt, rres_iseq, rres_mseq);
} else if (cnt == 0) {
if (print) printf ("(no data)\n");
} else {
printf ("%s ERROR %d\n", opname, cnt);
abort ();
}
for (int i = 0; i < cnt; i++)
{
const int is = (rres_iseq[i].instance_state == DDS_ALIVE_INSTANCE_STATE) ? 2 : (rres_iseq[i].instance_state == DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE) ? 1 : 0;
const int x = (rres_iseq[i].sample_state == DDS_NOT_READ_SAMPLE_STATE) + 2 * (rres_iseq[i].view_state == DDS_NEW_VIEW_STATE) + 4 * is;
states_seen[x][rres_iseq[i].valid_data]++;
/* invalid samples are expected to be zero except for the key fields */
if (!rres_iseq[i].valid_data)
{
if (rres_mseq[i].x != 0 || rres_mseq[i].y != 0 || rres_mseq[i].s != NULL)
abort ();
}
}
/* all returned data must match cond */
if (cond)
{
for (int i = 0; i < cnt; i++)
{
switch (cond->m_sample_states)
{
case DDS_SST_READ:
if (rres_iseq[i].sample_state != DDS_READ_SAMPLE_STATE) abort ();
break;
case DDS_SST_NOT_READ:
if (rres_iseq[i].sample_state != DDS_NOT_READ_SAMPLE_STATE) abort ();
break;
}
switch (cond->m_view_states)
{
case DDS_VST_NEW:
if (rres_iseq[i].view_state != DDS_NEW_VIEW_STATE) abort ();
break;
case DDS_VST_OLD:
if (rres_iseq[i].view_state != DDS_NOT_NEW_VIEW_STATE) abort ();
break;
}
switch (cond->m_instance_states)
{
case DDS_IST_ALIVE:
if (rres_iseq[i].instance_state != DDS_ALIVE_INSTANCE_STATE) abort ();
break;
case DDS_IST_NOT_ALIVE_NO_WRITERS:
if (rres_iseq[i].instance_state != DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE) abort ();
break;
case DDS_IST_NOT_ALIVE_DISPOSED:
if (rres_iseq[i].instance_state != DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE) abort ();
break;
case DDS_IST_NOT_ALIVE_NO_WRITERS | DDS_IST_NOT_ALIVE_DISPOSED:
if (rres_iseq[i].instance_state != DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE && rres_iseq[i].instance_state != DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE)
abort ();
break;
case DDS_IST_ALIVE | DDS_IST_NOT_ALIVE_NO_WRITERS:
if (rres_iseq[i].instance_state != DDS_ALIVE_INSTANCE_STATE && rres_iseq[i].instance_state != DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE)
abort ();
break;
case DDS_IST_ALIVE | DDS_IST_NOT_ALIVE_DISPOSED:
if (rres_iseq[i].instance_state != DDS_ALIVE_INSTANCE_STATE && rres_iseq[i].instance_state != DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE)
abort ();
break;
}
if (cond->m_query.m_filter)
{
/* invalid samples don't get the attributes zero'd out in the result, though the keys are guaranteed to be set; maybe I should change that and guarantee that the fields are 0 ... */
if (!cond->m_query.m_filter (&rres_mseq[i]))
abort ();
}
}
}
if (chk)
{
docheck (cnt, rres_iseq, rres_mseq, chk);
}
}
static void rdall (struct rhc *rhc, const struct check *chk, bool print, uint32_t states_seen[STATIC_ARRAY_DIM 2*2*3][2])
{
rdtkcond (rhc, NULL, chk, print, 0, "READ ALL", dds_rhc_read, states_seen);
}
static void tkall (struct rhc *rhc, const struct check *chk, bool print, uint32_t states_seen[STATIC_ARRAY_DIM 2*2*3][2])
{
rdtkcond (rhc, NULL, chk, print, 0, "TAKE ALL", dds_rhc_take, states_seen);
}
static void print_condmask (char *buf, size_t bufsz, const dds_readcond *cond)
{
size_t pos = 0;
const char *sep = "";
pos += (size_t) snprintf (buf + pos, bufsz - pos, "[");
switch (cond->m_sample_states)
{
case DDS_SST_READ:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sREAD", sep);
sep = ", ";
break;
case DDS_SST_NOT_READ:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sNOT_READ", sep);
sep = ", ";
break;
}
switch (cond->m_view_states)
{
case DDS_VST_NEW:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sNEW", sep);
sep = ", ";
break;
case DDS_VST_OLD:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sOLD", sep);
sep = ", ";
break;
}
switch (cond->m_instance_states)
{
case DDS_IST_ALIVE:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sALIVE", sep);
break;
case DDS_IST_NOT_ALIVE_NO_WRITERS:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sNO_WRITERS", sep);
break;
case DDS_IST_NOT_ALIVE_DISPOSED:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sDISPOSED", sep);
break;
case DDS_IST_NOT_ALIVE_NO_WRITERS | DDS_IST_NOT_ALIVE_DISPOSED:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sNOT_ALIVE", sep);
break;
case DDS_IST_ALIVE | DDS_IST_NOT_ALIVE_NO_WRITERS:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sALIVE | NO_WRITERS", sep);
break;
case DDS_IST_ALIVE | DDS_IST_NOT_ALIVE_DISPOSED:
pos += (size_t) snprintf (buf + pos, bufsz - pos, "%sALIVE | DISPOSED", sep);
break;
}
snprintf (buf + pos, bufsz - pos, "]");
}
static void rdcond (struct rhc *rhc, dds_readcond *cond, const struct check *chk, int max, bool print, uint32_t states_seen[STATIC_ARRAY_DIM 2*2*3][2])
{
char buf[100];
int pos;
pos = snprintf (buf, sizeof (buf), "READ COND %p %d ", (void *) cond, max);
print_condmask (buf + pos, sizeof (buf) - (size_t) pos, cond);
rdtkcond (rhc, cond, chk, print, max, buf, dds_rhc_read, states_seen);
}
static void tkcond (struct rhc *rhc, dds_readcond *cond, const struct check *chk, int max, bool print, uint32_t states_seen[STATIC_ARRAY_DIM 2*2*3][2])
{
char buf[100];
int pos;
pos = snprintf (buf, sizeof (buf), "TAKE COND %p %d ", (void *) cond, max);
print_condmask (buf + pos, sizeof (buf) - (size_t) pos, cond);
rdtkcond (rhc, cond, chk, print, max, buf, dds_rhc_take, states_seen);
}
static void wait_gc_cycle_impl (struct gcreq *gcreq)
{
os_mutexLock (&wait_gc_cycle_lock);
wait_gc_cycle_trig = 1;
os_condBroadcast (&wait_gc_cycle_cond);
os_mutexUnlock (&wait_gc_cycle_lock);
gcreq_free (gcreq);
}
static void wait_gc_cycle (void)
{
/* only single-threaded for now */
struct gcreq *gcreq = gcreq_new (gv.gcreq_queue, wait_gc_cycle_impl);
#ifndef NDEBUG
os_mutexLock (&wait_gc_cycle_lock);
assert (wait_gc_cycle_trig == 0);
os_mutexUnlock (&wait_gc_cycle_lock);
#endif
gcreq_enqueue (gcreq);
os_mutexLock (&wait_gc_cycle_lock);
while (!wait_gc_cycle_trig)
os_condWait (&wait_gc_cycle_cond, &wait_gc_cycle_lock);
wait_gc_cycle_trig = 0;
os_mutexUnlock (&wait_gc_cycle_lock);
}
static bool qcpred_key (const void *vx)
{
const RhcTypes_T *x = vx;
return (x->k % 2) == 0;
}
static bool qcpred_attr2 (const void *vx)
{
const RhcTypes_T *x = vx;
return (x->x % 2) == 0;
}
static bool qcpred_attr3 (const void *vx)
{
const RhcTypes_T *x = vx;
return (x->x % 3) == 0;
}
static dds_readcond *get_condaddr (dds_entity_t x)
{
struct dds_entity *e;
if (dds_entity_lock (x, DDS_KIND_DONTCARE, &e) < 0)
abort();
assert (dds_entity_kind (e) == DDS_KIND_COND_READ || dds_entity_kind (e) == DDS_KIND_COND_QUERY);
dds_entity_unlock (e);
return (dds_readcond *) e;
}
static void print_cond_w_addr (const char *label, dds_entity_t x)
{
char buf[100];
struct dds_entity *e;
if (dds_entity_lock (x, DDS_KIND_DONTCARE, &e) < 0)
abort();
assert (dds_entity_kind (e) == DDS_KIND_COND_READ || dds_entity_kind (e) == DDS_KIND_COND_QUERY);
print_condmask (buf, sizeof (buf), (dds_readcond *) e);
printf ("%s: %"PRIu32" => %p %s\n", label, x, (void *) e, buf);
dds_entity_unlock (e);
}
static dds_entity_t readcond_wrapper (dds_entity_t reader, uint32_t mask, dds_querycondition_filter_fn filter)
{
(void) filter;
return dds_create_readcondition (reader, mask);
}
static void test_conditions (dds_entity_t pp, dds_entity_t tp, const int count, dds_entity_t (*create_cond) (dds_entity_t reader, uint32_t mask, dds_querycondition_filter_fn filter), dds_querycondition_filter_fn filter0, dds_querycondition_filter_fn filter1, bool print)
{
dds_qos_t *qos = dds_create_qos ();
dds_qset_history (qos, DDS_HISTORY_KEEP_LAST, MAX_HIST_DEPTH);
dds_qset_destination_order (qos, DDS_DESTINATIONORDER_BY_SOURCE_TIMESTAMP);
/* two identical readers because we need 63 conditions while we can currently only attach 32 a single reader */
dds_entity_t rd[] = { dds_create_reader (pp, tp, qos, NULL), dds_create_reader (pp, tp, qos, NULL) };
const size_t nrd = sizeof (rd) / sizeof (rd[0]);
dds_delete_qos (qos);
struct rhc *rhc[sizeof (rd) / sizeof (rd[0])];
for (size_t i = 0; i < sizeof (rd) / sizeof (rd[0]); i++)
{
struct dds_entity *x;
if (dds_entity_lock (rd[i], DDS_KIND_READER, &x) < 0)
abort ();
dds_reader *rdp = (dds_reader *) x;
rhc[i] = rdp->m_rd->rhc;
dds_entity_unlock (x);
}
struct proxy_writer *wr[] = { mkwr (0), mkwr (1), mkwr (1) };
static const uint32_t stab[] = {
DDS_READ_SAMPLE_STATE, DDS_NOT_READ_SAMPLE_STATE,
DDS_READ_SAMPLE_STATE | DDS_NOT_READ_SAMPLE_STATE
};
const int nstab = (int) (sizeof (stab) / sizeof (stab[0]));
static const uint32_t vtab[] = {
DDS_NEW_VIEW_STATE, DDS_NOT_NEW_VIEW_STATE,
DDS_NEW_VIEW_STATE | DDS_NOT_NEW_VIEW_STATE
};
const int nvtab = (int) (sizeof (vtab) / sizeof (vtab[0]));
static const uint32_t itab[] = {
DDS_ALIVE_INSTANCE_STATE, DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE,
DDS_ALIVE_INSTANCE_STATE | DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE,
DDS_ALIVE_INSTANCE_STATE | DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE,
DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE | DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE,
DDS_ALIVE_INSTANCE_STATE | DDS_NOT_ALIVE_NO_WRITERS_INSTANCE_STATE | DDS_NOT_ALIVE_DISPOSED_INSTANCE_STATE
};
const int nitab = (int) (sizeof (itab) / sizeof (itab[0]));
const int nconds = nstab * nvtab * nitab;
dds_entity_t gdcond = dds_create_guardcondition (pp);
dds_entity_t waitset = dds_create_waitset(pp);
dds_waitset_attach(waitset, gdcond, 888);
/* create a conditions for every possible state mask */
assert (nconds == 63);
dds_entity_t conds[63];
dds_readcond *rhcconds[63];
{
int ci = 0;
for (int s = 0; s < nstab; s++)
for (int v = 0; v < nvtab; v++)
for (int i = 0; i < nitab; i++)
{
assert (ci / 32 < (int) (sizeof (rd) / sizeof (rd[0])));
conds[ci] = create_cond (rd[ci / 32], stab[s] | vtab[v] | itab[i], ((ci % 2) == 0) ? filter0 : filter1);
if (conds[ci] <= 0) abort ();
rhcconds[ci] = get_condaddr (conds[ci]);
if (print) {
char buf[10];
snprintf (buf, sizeof (buf), "conds[%d]", ci);
print_cond_w_addr (buf, conds[ci]);
}
dds_waitset_attach(waitset, conds[ci], ci);
ci++;
}
}
/* simply sanity check on the guard condition and waitset triggering */
{
bool v;
int n;
dds_attach_t xs[2];
dds_read_guardcondition (gdcond, &v);
assert (!v);
dds_set_guardcondition (gdcond, true);
n = dds_waitset_wait (waitset, xs, sizeof(xs) / sizeof(xs[0]), 0);
assert (n == 1);
(void)n;
assert (xs[0] == 888);
dds_read_guardcondition (gdcond, &v);
assert (v);
dds_take_guardcondition (gdcond, &v);
assert (v);
dds_read_guardcondition (gdcond, &v);
assert (!v);
n = dds_waitset_wait (waitset, xs, sizeof(xs) / sizeof(xs[0]), 0);
assert (n == 0);
(void)n;
}
/* relative frequency table of operations: */
static const char *operstr[] = {
[0] = "w",
[1] = "wd",
[2] = "d",
[3] = "u",
[4] = "du",
[5] = "wdu",
[6] = "rdall",
[7] = "tkall",
[8] = "rdc",
[9] = "tkc",
[10] = "tkc1",
[11] = "delwr"
};
static const uint32_t opfreqs[] = {
[0] = 500, /* write */
[1] = 100, /* variants with dispose & unregister */
[2] = 100,
[3] = 300, /* just unregister */
[4] = 100, /* variants with dispose & unregister */
[5] = 100,
[6] = 50, /* read all */
[7] = 5, /* take all */
[8] = 200, /* read cond */
[9] = 30, /* take cond */
[10] = 100, /* take cond, max 1 */
[11] = 1 /* unreg writer */
};
uint32_t opthres[sizeof (opfreqs) / sizeof (opfreqs[0])];
{
const size_t n = sizeof (opfreqs) / sizeof (opfreqs[0]);
uint32_t sum = 0;
for (size_t i = 0; i < n; i++)
sum += opfreqs[i];
const uint32_t scale = UINT32_MAX / sum;
sum = 0;
for (size_t i = 0; i < n; i++)
{
sum += opfreqs[i];
opthres[i] = sum * scale;
}
}
uint32_t states_seen[2 * 2 * 3][2] = {{ 0 }};
uint32_t opcount[sizeof (opfreqs) / sizeof (opfreqs[0])] = { 0 };
int lastprint_pct = 0;
for (int i = 0; i < count; i++)
{
const int32_t keyval = (int32_t) (genrand_int32 () % N_KEYVALS);
const uint32_t which = genrand_int32 () % 3;
uint32_t oper_base;
uint32_t oper;
/* generate uniform number in range 0 .. N, then map to operation following the frequency table */
do {
oper_base = genrand_int32 ();
} while (oper_base >= opthres[sizeof (opfreqs) / sizeof (opfreqs[0]) - 1]);
for (oper = 0; oper < sizeof (opfreqs) / sizeof (opfreqs[0]); oper++)
{
if (oper_base < opthres[oper])
break;
}
opcount[oper]++;
if (100 * i / count > lastprint_pct)
{
lastprint_pct = 100 * i / count;
printf ("%d%%%c", lastprint_pct, print ? '\n' : '\r');
fflush (stdout);
}
switch (oper)
{
case 0: { /* wr */
struct ddsi_serdata *s = mksample (keyval, 0);
for (size_t k = 0; k < nrd; k++)
store (rhc[k], wr[which], ddsi_serdata_ref (s), print && k == 0);
ddsi_serdata_unref (s);
break;
}
case 1: { /* wr disp */
struct ddsi_serdata *s = mksample (keyval, NN_STATUSINFO_DISPOSE);
for (size_t k = 0; k < nrd; k++)
store (rhc[k], wr[which], ddsi_serdata_ref (s), print && k == 0);
ddsi_serdata_unref (s);
break;
}
case 2: { /* disp */
struct ddsi_serdata *s = mkkeysample (keyval, NN_STATUSINFO_DISPOSE);
for (size_t k = 0; k < nrd; k++)
store (rhc[k], wr[which], ddsi_serdata_ref (s), print && k == 0);
ddsi_serdata_unref (s);
break;
}
case 3: { /* unreg */
struct ddsi_serdata *s = mkkeysample (keyval, NN_STATUSINFO_UNREGISTER);
for (size_t k = 0; k < nrd; k++)
store (rhc[k], wr[which], ddsi_serdata_ref (s), print && k == 0);
ddsi_serdata_unref (s);
break;
}
case 4: { /* disp unreg */
struct ddsi_serdata *s = mkkeysample (keyval, NN_STATUSINFO_DISPOSE | NN_STATUSINFO_UNREGISTER);
for (size_t k = 0; k < nrd; k++)
store (rhc[k], wr[which], ddsi_serdata_ref (s), print && k == 0);
ddsi_serdata_unref (s);
break;
}
case 5: { /* wr disp unreg */
struct ddsi_serdata *s = mksample (keyval, NN_STATUSINFO_DISPOSE | NN_STATUSINFO_UNREGISTER);
for (size_t k = 0; k < nrd; k++)
store (rhc[k], wr[which], ddsi_serdata_ref (s), print && k == 0);
ddsi_serdata_unref (s);
break;
}
case 6:
for (size_t k = 0; k < nrd; k++)
rdall (rhc[k], NULL, print && k == 0, states_seen);
break;
case 7:
for (size_t k = 0; k < nrd; k++)
tkall (rhc[k], NULL, print && k == 0, states_seen);
break;
case 8: {
uint32_t cond = genrand_int32 () % (uint32_t) nconds;
for (size_t k = 0; k < nrd; k++)
rdcond (rhc[k], rhcconds[cond], NULL, 0, print && k == 0, states_seen);
break;
}
case 9: {
uint32_t cond = genrand_int32 () % (uint32_t) nconds;
for (size_t k = 0; k < nrd; k++)
tkcond (rhc[k], rhcconds[cond], NULL, 0, print && k == 0, states_seen);
break;
}
case 10: {
uint32_t cond = genrand_int32 () % (uint32_t) nconds;
for (size_t k = 0; k < nrd; k++)
tkcond (rhc[k], rhcconds[cond], NULL, 1, print && k == 0, states_seen);
break;
}
case 11:
thread_state_awake (mainthread);
struct proxy_writer_info wr_info;
wr_info.auto_dispose = wr[which]->c.xqos->writer_data_lifecycle.autodispose_unregistered_instances;
wr_info.guid = wr[which]->e.guid;
wr_info.iid = wr[which]->e.iid;
wr_info.ownership_strength = wr[which]->c.xqos->ownership_strength.value;
for (size_t k = 0; k < nrd; k++)
dds_rhc_unregister_wr (rhc[k], &wr_info);
thread_state_asleep (mainthread);
break;
}
if ((i % 200) == 0)
wait_gc_cycle ();
}
for (size_t oper = 0; oper < sizeof (opcount) / sizeof (opcount[0]); oper++)
printf ("%5s: %8"PRIu32"\n", operstr[oper], opcount[oper]);
for (int i = 0; i < (int) (sizeof (states_seen) / sizeof (states_seen[0])); i++)
{
const char sst = (i & 1) ? 'N' : 'R';
const char vst = (i & 2) ? 'N' : 'O';
const char ist = (i >> 2) == 2 ? 'A' : (i >> 2) == 1 ? 'D' : 'U';
printf ("%c%c%c: invalid %8"PRIu32" valid %8"PRIu32"\n", sst, vst, ist, states_seen[i][0], states_seen[i][1]);
}
dds_waitset_detach (waitset, gdcond);
for (int ci = 0; ci < nconds; ci++)
dds_waitset_detach (waitset, conds[ci]);
dds_delete (waitset);
dds_delete (gdcond);
for (int ci = 0; ci < nconds; ci++)
dds_delete (conds[ci]);
for (size_t i = 0; i < nrd; i++)
dds_delete (rd[i]);
for (size_t i = 0; i < sizeof (wr) / sizeof (wr[0]); i++)
fwr (wr[i]);
}
int main (int argc, char **argv)
{
dds_entity_t pp = dds_create_participant(DDS_DOMAIN_DEFAULT, NULL, NULL);
dds_entity_t tp = dds_create_topic(pp, &RhcTypes_T_desc, "RhcTypes_T", NULL, NULL);
uint32_t states_seen[2 * 2 * 3][2] = {{ 0 }};
unsigned seed = 0;
bool print = false;
int first = 0, count = 10000;
os_mutexInit (&wait_gc_cycle_lock);
os_condInit (&wait_gc_cycle_cond, &wait_gc_cycle_lock);
if (argc > 1)
seed = (unsigned) atoi (argv[1]);
if (seed == 0)
seed = (unsigned) os_getpid ();
if (argc > 2)
first = atoi (argv[2]);
if (argc > 3)
count = atoi (argv[3]);
if (argc > 4)
print = (atoi (argv[4]) != 0);
printf ("prng seed %u first %d count %d print %d\n", seed, first, count, print);
init_genrand (seed);
memset (rres_mseq, 0, sizeof (rres_mseq));
for (size_t i = 0; i < sizeof (rres_iseq) / sizeof(rres_iseq[0]); i++)
rres_ptrs[i] = &rres_mseq[i];
tref_dds = dds_time();
mainthread = lookup_thread_state ();
{
struct dds_entity *x;
if (dds_entity_lock(tp, DDS_KIND_TOPIC, &x) < 0) abort();
mdtopic = dds_topic_lookup(x->m_domain, "RhcTypes_T");
dds_entity_unlock(x);
}
if (0 >= first)
{
if (print)
printf ("************* 0 *************\n");
struct rhc *rhc = mkrhc (NULL, NN_KEEP_LAST_HISTORY_QOS, 1, NN_BY_SOURCE_TIMESTAMP_DESTINATIONORDER_QOS);
struct proxy_writer *wr0 = mkwr (1);
uint64_t iid0, iid1, iid_t;
iid0 = store (rhc, wr0, mksample (0, 0), print);
iid1 = store (rhc, wr0, mksample (1, NN_STATUSINFO_DISPOSE), print);
const struct check c0[] = {
{ "NNA", iid0, wr0->e.iid, 0,0, 1, 0,1 },
{ "NND", iid1, wr0->e.iid, 0,0, 1, 1,2 },
{ 0, 0, 0, 0, 0, 0, 0, 0 }
};
rdall (rhc, c0, print, states_seen);
iid_t = store (rhc, wr0, mkkeysample (0, NN_STATUSINFO_UNREGISTER), print);
assert (iid_t == iid0);
(void)iid0;
(void)iid_t;
const struct check c1[] = {
{ "ROU", iid0, wr0->e.iid, 0,0, 1, 0,1 },
{ "NOU", iid0, 0, 0,0, 0, 0,0 },
{ "ROD", iid1, wr0->e.iid, 0,0, 1, 1,2 },
{ 0, 0, 0, 0, 0, 0, 0, 0 }
};
rdall (rhc, c1, print, states_seen);
thread_state_awake (mainthread);
struct proxy_writer_info wr0_info;
wr0_info.auto_dispose = wr0->c.xqos->writer_data_lifecycle.autodispose_unregistered_instances;
wr0_info.guid = wr0->e.guid;
wr0_info.iid = wr0->e.iid;
wr0_info.ownership_strength = wr0->c.xqos->ownership_strength.value;
dds_rhc_unregister_wr (rhc, &wr0_info);
thread_state_asleep (mainthread);
const struct check c2[] = {
{ "ROU", iid0, wr0->e.iid, 0,0, 1, 0,1 },
{ "ROU", iid0, 0, 0,0, 0, 0,0 },
{ "ROD", iid1, wr0->e.iid, 0,0, 1, 1,2 },
{ "NOD", iid1, 0, 0,0, 0, 1,0 },
{ 0, 0, 0, 0, 0, 0, 0, 0 }
};
tkall (rhc, c2, print, states_seen);
frhc (rhc);
fwr (wr0);
}
if (1 >= first)
{
if (print)
printf ("************* 1 *************\n");
struct rhc *rhc = mkrhc (NULL, NN_KEEP_LAST_HISTORY_QOS, 4, NN_BY_SOURCE_TIMESTAMP_DESTINATIONORDER_QOS);
struct proxy_writer *wr[] = { mkwr (0), mkwr (0), mkwr (0) };
uint64_t iid0, iid_t;
int nregs = 3, isreg[] = { 1, 1, 1 };
iid0 = store (rhc, wr[0], mksample (0, 0), print);
iid_t = store (rhc, wr[1], mksample (0, 0), print); assert (iid0 == iid_t);
iid_t = store (rhc, wr[2], mksample (0, 0), print); assert (iid0 == iid_t);
(void)iid0;
tkall (rhc, NULL, print, states_seen);
for (int i = 0; i < 3*3 * 3*3 * 3*3 * 3*3; i++)
{
for (int pos = 0, base = 1; pos < 3; pos++, base *= 3*3)
{
int which = (((i / base) / 3) + pos) % 3;
int oper = (i / base) % 3;
switch (oper)
{
case 0:
iid_t = store (rhc, wr[which], mksample (0, 0), print);
if (!isreg[which]) { nregs++; isreg[which] = 1; }
break;
case 1:
iid_t = store (rhc, wr[which], mkkeysample (0, NN_STATUSINFO_DISPOSE), print);
if (!isreg[which]) { nregs++; isreg[which] = 1; }
break;
case 2:
if (nregs > 1 || !isreg[which])
{
iid_t = store (rhc, wr[which], mkkeysample (0, NN_STATUSINFO_UNREGISTER), print);
if (isreg[which]) { isreg[which] = 0; nregs--; }
}
break;
}
assert (iid_t == iid0);
}
}
tkall (rhc, 0, print, states_seen);
wait_gc_cycle ();
assert (nregs > 0);
for (int i = 0; i < 3; i++)
{
if (isreg[i])
{
iid_t = store (rhc, wr[i], mkkeysample (0, NN_STATUSINFO_UNREGISTER), print);
assert (iid_t == iid0);
isreg[i] = 0;
nregs--;
}
}
assert (nregs == 0);
tkall (rhc, 0, print, states_seen);
wait_gc_cycle ();
iid_t = store (rhc, wr[0], mksample (0, 0), print);
assert (iid_t != iid0);
iid0 = iid_t;
iid_t = store (rhc, wr[0], mkkeysample (0, NN_STATUSINFO_UNREGISTER), print);
assert (iid_t == iid0);
frhc (rhc);
for (size_t i = 0; i < sizeof (wr) / sizeof (wr[0]); i++)
fwr (wr[i]);
}
{
static const struct {
dds_entity_t (*create) (dds_entity_t, uint32_t, dds_querycondition_filter_fn);
dds_querycondition_filter_fn filter0;
dds_querycondition_filter_fn filter1;
} zztab[] = {
{ readcond_wrapper, 0, 0 },
{ dds_create_querycondition, qcpred_key, qcpred_attr2 },
{ dds_create_querycondition, qcpred_attr2, qcpred_attr3 }
};
for (int zz = 0; zz < (int) (sizeof (zztab) / sizeof (zztab[0])); zz++)
if (zz + 2 >= first)
{
if (print)
printf ("************* %d *************\n", zz + 2);
test_conditions (pp, tp, count, zztab[zz].create, zztab[zz].filter0, zztab[zz].filter1, print);
}
}
os_condDestroy (&wait_gc_cycle_cond);
os_mutexDestroy (&wait_gc_cycle_lock);
for (size_t i = 0; i < sizeof (rres_iseq) / sizeof (rres_iseq[0]); i++)
RhcTypes_T_free (&rres_mseq[i], DDS_FREE_CONTENTS);
dds_delete(pp);
return 0;
}

2
src/os/include/os/os_log.h
View file

@ -69,6 +69,8 @@ extern "C" {
/** Debug/trace messages related to throttling. */
#define DDS_LC_THROTTLE (32768u)
/** All common trace categories. */
#define DDS_LC_RHC (65536u)
/** All common trace categories. */
#define DDS_LC_ALL \
(DDS_LC_FATAL | DDS_LC_ERROR | DDS_LC_WARNING | DDS_LC_INFO | \
DDS_LC_CONFIG | DDS_LC_DISCOVERY | DDS_LC_DATA | DDS_LC_TRACE | \