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

Abstract RHC interface

Browse source 
This makes it possible to use a different RHC implementations for
different readers and removes the need for the RHC interface to be part
of the global state.

Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson 2019-06-27 09:51:12 +02:00 • committed by eboasson
parent 483f4d2b77
commit 2e9ce9b4c1
20 changed files with 3047 additions and 2872 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/core/ddsc/CMakeLists.txt
View file

@ -19,6 +19,7 @@ PREPEND(srcs_ddsc "${CMAKE_CURRENT_LIST_DIR}/src"
dds_init.c
dds_publisher.c
dds_rhc.c
dds_rhc_default.c
dds_domain.c
dds_instance.c
dds_qos.c
@ -67,6 +68,7 @@ PREPEND(hdrs_private_ddsc "${CMAKE_CURRENT_LIST_DIR}/src"
dds__guardcond.h
dds__reader.h
dds__rhc.h
dds__rhc_default.h
dds__stream.h
dds__subscriber.h
dds__topic.h

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

@ -12,60 +12,81 @@
#ifndef _DDS_RHC_H_
#define _DDS_RHC_H_
#include "dds/ddsrt/static_assert.h"
#include "dds/ddsi/q_rhc.h"
#include "dds__types.h" /* for dds_readcond */
#define NO_STATE_MASK_SET (DDS_ANY_STATE + 1)
#if defined (__cplusplus)
extern "C" {
#endif
struct rhc;
struct dds_qos;
struct ddsi_serdata;
struct ddsi_tkmap_instance;
struct proxy_writer_info;
struct dds_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);
typedef int (*dds_rhc_read_t) (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond);
typedef int (*dds_rhc_take_t) (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond);
typedef int (*dds_rhc_takecdr_t) (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle);
DDS_EXPORT uint32_t dds_rhc_lock_samples (struct rhc *rhc);
typedef bool (*dds_rhc_add_readcondition_t) (struct dds_readcond *cond);
typedef void (*dds_rhc_remove_readcondition_t) (struct dds_readcond *cond);
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);
typedef uint32_t (*dds_rhc_lock_samples_t) (struct dds_rhc *rhc);
DDS_EXPORT int
dds_rhc_read(
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);
DDS_EXPORT int
dds_rhc_take(
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);
struct dds_rhc_ops {
/* A copy of DDSI rhc ops comes first so we can use either interface without
additional indirections */
struct rhc_ops rhc_ops;
dds_rhc_read_t read;
dds_rhc_take_t take;
dds_rhc_takecdr_t takecdr;
dds_rhc_add_readcondition_t add_readcondition;
dds_rhc_remove_readcondition_t remove_readcondition;
dds_rhc_lock_samples_t lock_samples;
};
DDS_EXPORT void dds_rhc_set_qos (struct rhc * rhc, const struct dds_qos * qos);
struct dds_rhc {
union {
const struct dds_rhc_ops *ops;
struct rhc rhc;
} common;
};
DDS_EXPORT bool dds_rhc_add_readcondition (dds_readcond * cond);
DDS_EXPORT void dds_rhc_remove_readcondition (dds_readcond * cond);
DDSRT_STATIC_ASSERT (offsetof (struct dds_rhc, common.ops) == offsetof (struct rhc, ops));
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, uint32_t sample_states,
uint32_t view_states, uint32_t instance_states,
dds_instance_handle_t handle
);
DDS_EXPORT inline bool dds_rhc_store (struct dds_rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info, struct ddsi_serdata * __restrict sample, struct ddsi_tkmap_instance * __restrict tk) {
return rhc->common.ops->rhc_ops.store (&rhc->common.rhc, pwr_info, sample, tk);
}
DDS_EXPORT inline void dds_rhc_unregister_wr (struct dds_rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info) {
rhc->common.ops->rhc_ops.unregister_wr (&rhc->common.rhc, pwr_info);
}
DDS_EXPORT inline void dds_rhc_relinquish_ownership (struct dds_rhc * __restrict rhc, const uint64_t wr_iid) {
rhc->common.ops->rhc_ops.relinquish_ownership (&rhc->common.rhc, wr_iid);
}
DDS_EXPORT inline void dds_rhc_set_qos (struct dds_rhc *rhc, const struct dds_qos *qos) {
rhc->common.ops->rhc_ops.set_qos (&rhc->common.rhc, qos);
}
DDS_EXPORT inline void dds_rhc_free (struct dds_rhc *rhc) {
rhc->common.ops->rhc_ops.free (&rhc->common.rhc);
}
DDS_EXPORT inline int dds_rhc_read (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond) {
return rhc->common.ops->read (rhc, lock, values, info_seq, max_samples, mask, handle, cond);
}
DDS_EXPORT inline int dds_rhc_take (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond) {
return rhc->common.ops->take (rhc, lock, values, info_seq, max_samples, mask, handle, cond);
}
DDS_EXPORT inline int dds_rhc_takecdr (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle) {
return rhc->common.ops->takecdr (rhc, lock, values, info_seq, max_samples, sample_states, view_states, instance_states, handle);
}
DDS_EXPORT inline bool dds_rhc_add_readcondition (struct dds_readcond *cond) {
return cond->m_rhc->common.ops->add_readcondition (cond);
}
DDS_EXPORT inline void dds_rhc_remove_readcondition (struct dds_readcond *cond) {
cond->m_rhc->common.ops->remove_readcondition (cond);
}
DDS_EXPORT inline uint32_t dds_rhc_lock_samples (struct dds_rhc *rhc) {
return rhc->common.ops->lock_samples (rhc);
}
#if defined (__cplusplus)
}

28
src/core/ddsc/src/dds__rhc_default.h Normal file
View file

@ -0,0 +1,28 @@
/*
* 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
*/
#ifndef _DDS_RHC_DEFAULT_H_
#define _DDS_RHC_DEFAULT_H_
#if defined (__cplusplus)
extern "C" {
#endif
struct dds_rhc;
struct dds_reader;
struct ddsi_sertopic;
DDS_EXPORT struct dds_rhc *dds_rhc_default_new (struct dds_reader *reader, const struct ddsi_sertopic *topic);
#if defined (__cplusplus)
}
#endif
#endif

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

@ -211,6 +211,7 @@ typedef struct dds_participant {
typedef struct dds_reader {
struct dds_entity m_entity;
const struct dds_topic *m_topic;
struct dds_rhc *m_rhc; /* aliases m_rd->rhc with a wider interface, FIXME: but m_rd owns it for resource management */
struct reader *m_rd;
bool m_data_on_readers;
bool m_loan_out;
@ -263,7 +264,7 @@ typedef uint32_t dds_querycond_mask_t;
typedef struct dds_readcond {
dds_entity m_entity;
struct rhc *m_rhc;
struct dds_rhc *m_rhc;
uint32_t m_qminv;
uint32_t m_sample_states;
uint32_t m_view_states;

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

@ -1,5 +1,5 @@
/*
* Copyright(c) 2006 to 2018 ADLINK Technology Limited and others
* Copyright(c) 2006 to 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

6
src/core/ddsc/src/dds_init.c
View file

@ -272,12 +272,6 @@ void ddsi_plugin_init (void)
ddsi_plugin.builtintopic_is_visible = dds__builtin_is_visible;
ddsi_plugin.builtintopic_get_tkmap_entry = dds__builtin_get_tkmap_entry;
ddsi_plugin.builtintopic_write = dds__builtin_write;
ddsi_plugin.rhc_plugin.rhc_free_fn = dds_rhc_free;
ddsi_plugin.rhc_plugin.rhc_store_fn = dds_rhc_store;
ddsi_plugin.rhc_plugin.rhc_unregister_wr_fn = dds_rhc_unregister_wr;
ddsi_plugin.rhc_plugin.rhc_relinquish_ownership_fn = dds_rhc_relinquish_ownership;
ddsi_plugin.rhc_plugin.rhc_set_qos_fn = dds_rhc_set_qos;
}
//provides explicit default domain id.

6
src/core/ddsc/src/dds_read.c
View file

@ -117,9 +117,9 @@ static dds_return_t dds_read_impl (bool take, dds_entity_t reader_or_condition,
dds_entity_status_reset (rd->m_entity.m_parent, DDS_DATA_ON_READERS_STATUS);
if (take)
ret = dds_rhc_take (rd->m_rd->rhc, lock, buf, si, maxs, mask, hand, cond);
ret = dds_rhc_take (rd->m_rhc, lock, buf, si, maxs, mask, hand, cond);
else
ret = dds_rhc_read (rd->m_rd->rhc, lock, buf, si, maxs, mask, hand, cond);
ret = dds_rhc_read (rd->m_rhc, lock, buf, si, maxs, mask, hand, cond);
/* if no data read, restore the state to what it was before the call, with the sole
exception of holding on to a buffer we just allocated and that is pointed to by
@ -186,7 +186,7 @@ static dds_return_t dds_readcdr_impl (bool take, dds_entity_t reader_or_conditio
assert (dds_entity_kind (rd->m_entity.m_parent) == DDS_KIND_SUBSCRIBER);
dds_entity_status_reset (rd->m_entity.m_parent, DDS_DATA_ON_READERS_STATUS);
ret = dds_rhc_takecdr (rd->m_rd->rhc, lock, buf, si, maxs, mask & DDS_ANY_SAMPLE_STATE, mask & DDS_ANY_VIEW_STATE, mask & DDS_ANY_INSTANCE_STATE, hand);
ret = dds_rhc_takecdr (rd->m_rhc, lock, buf, si, maxs, mask & DDS_ANY_SAMPLE_STATE, mask & DDS_ANY_VIEW_STATE, mask & DDS_ANY_INSTANCE_STATE, hand);
dds_entity_unpin (entity);
fail_awake:

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

@ -41,7 +41,7 @@ dds_readcond *dds_create_readcond (dds_reader *rd, dds_entity_kind_t kind, uint3
(void) dds_entity_init (&cond->m_entity, &rd->m_entity, kind, NULL, NULL, 0);
cond->m_entity.m_iid = ddsi_iid_gen ();
dds_entity_register_child (&rd->m_entity, &cond->m_entity);
cond->m_rhc = rd->m_rd->rhc;
cond->m_rhc = rd->m_rhc;
cond->m_sample_states = mask & DDS_ANY_SAMPLE_STATE;
cond->m_view_states = mask & DDS_ANY_VIEW_STATE;
cond->m_instance_states = mask & DDS_ANY_INSTANCE_STATE;

10
src/core/ddsc/src/dds_reader.c
View file

@ -19,6 +19,7 @@
#include "dds__listener.h"
#include "dds__init.h"
#include "dds__rhc.h"
#include "dds__rhc_default.h"
#include "dds__topic.h"
#include "dds__get_status.h"
#include "dds__qos.h"
@ -296,7 +297,6 @@ dds_entity_t dds_create_reader (dds_entity_t participant_or_subscriber, dds_enti
dds_subscriber *sub = NULL;
dds_entity_t subscriber;
dds_reader *rd;
struct rhc *rhc;
dds_topic *tp;
dds_entity_t reader;
dds_entity_t t;
@ -382,7 +382,7 @@ dds_entity_t dds_create_reader (dds_entity_t participant_or_subscriber, dds_enti
reader = dds_entity_init (&rd->m_entity, &sub->m_entity, DDS_KIND_READER, rqos, listener, DDS_READER_STATUS_MASK);
rd->m_sample_rejected_status.last_reason = DDS_NOT_REJECTED;
rd->m_topic = tp;
rhc = dds_rhc_new (rd, tp->m_stopic);
rd->m_rhc = dds_rhc_default_new (rd, tp->m_stopic);
dds_entity_add_ref_locked (&tp->m_entity);
/* Extra claim of this reader to make sure that the delete waits until DDSI
@ -393,7 +393,7 @@ dds_entity_t dds_create_reader (dds_entity_t participant_or_subscriber, dds_enti
ddsrt_mutex_unlock (&sub->m_entity.m_mutex);
thread_state_awake (lookup_thread_state ());
ret = new_reader (&rd->m_rd, &rd->m_entity.m_guid, NULL, &sub->m_entity.m_participant->m_guid, tp->m_stopic, rqos, rhc, dds_reader_status_cb, rd);
ret = new_reader (&rd->m_rd, &rd->m_entity.m_guid, NULL, &sub->m_entity.m_participant->m_guid, tp->m_stopic, rqos, &rd->m_rhc->common.rhc, dds_reader_status_cb, rd);
ddsrt_mutex_lock (&sub->m_entity.m_mutex);
ddsrt_mutex_lock (&tp->m_entity.m_mutex);
assert (ret == DDS_RETCODE_OK); /* FIXME: can be out-of-resources at the very least */
@ -404,7 +404,7 @@ dds_entity_t dds_create_reader (dds_entity_t participant_or_subscriber, dds_enti
/* For persistent data register reader with durability */
if (dds_global.m_dur_reader && (rd->m_entity.m_qos->durability.kind > DDS_DURABILITY_TRANSIENT_LOCAL)) {
(dds_global.m_dur_reader) (rd, rhc);
(dds_global.m_dur_reader) (rd, &rd->m_rhc->common.rhc);
}
dds_topic_unlock (tp);
dds_subscriber_unlock (sub);
@ -484,7 +484,7 @@ uint32_t dds_reader_lock_samples (dds_entity_t reader)
uint32_t n;
if (dds_reader_lock (reader, &rd) != DDS_RETCODE_OK)
return 0;
n = dds_rhc_lock_samples (rd->m_rd->rhc);
n = dds_rhc_lock_samples (rd->m_rhc);
dds_reader_unlock (rd);
return n;
}

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

File diff suppressed because it is too large Load diff

2808
src/core/ddsc/src/dds_rhc_default.c Normal file
View file

File diff suppressed because it is too large Load diff

3
src/core/ddsc/src/dds_write.c
View file

@ -16,6 +16,7 @@
#include "dds/ddsi/ddsi_tkmap.h"
#include "dds/ddsi/q_thread.h"
#include "dds/ddsi/q_xmsg.h"
#include "dds/ddsi/q_rhc.h"
#include "dds/ddsi/ddsi_serdata.h"
#include "dds__stream.h"
#include "dds/ddsi/q_transmit.h"
@ -71,7 +72,7 @@ dds_return_t dds_write_ts (dds_entity_t writer, const void *data, dds_time_t tim
static dds_return_t try_store (struct rhc *rhc, const struct proxy_writer_info *pwr_info, struct ddsi_serdata *payload, struct ddsi_tkmap_instance *tk, dds_duration_t *max_block_ms)
{
while (!(ddsi_plugin.rhc_plugin.rhc_store_fn) (rhc, pwr_info, payload, tk))
while (! rhc_store (rhc, pwr_info, payload, tk))
{
if (*max_block_ms > 0)
{

4
src/core/ddsi/CMakeLists.txt
View file

@ -22,7 +22,6 @@ PREPEND(srcs_ddsi "${CMAKE_CURRENT_LIST_DIR}/src"
ddsi_serdata_default.c
ddsi_sertopic.c
ddsi_sertopic_default.c
ddsi_rhc_plugin.c
ddsi_iid.c
ddsi_tkmap.c
ddsi_vendor.c
@ -54,6 +53,7 @@ PREPEND(srcs_ddsi "${CMAKE_CURRENT_LIST_DIR}/src"
q_transmit.c
q_inverse_uint32_set.c
q_whc.c
q_rhc.c
q_xevent.c
q_xmsg.c
q_freelist.c
@ -73,7 +73,6 @@ PREPEND(hdrs_private_ddsi "${CMAKE_CURRENT_LIST_DIR}/include/dds/ddsi"
ddsi_serdata.h
ddsi_sertopic.h
ddsi_serdata_default.h
ddsi_rhc_plugin.h
ddsi_iid.h
ddsi_tkmap.h
ddsi_vendor.h
@ -102,6 +101,7 @@ PREPEND(hdrs_private_ddsi "${CMAKE_CURRENT_LIST_DIR}/include/dds/ddsi"
q_qosmatch.h
q_radmin.h
q_receive.h
q_rhc.h
q_rtps.h
q_security.h
q_sockwaitset.h

53
src/core/ddsi/include/dds/ddsi/ddsi_rhc_plugin.h
View file

@ -1,53 +0,0 @@
/*
* 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
*/
#ifndef DDSI_RHC_PLUGIN_H
#define DDSI_RHC_PLUGIN_H
#if defined (__cplusplus)
extern "C" {
#endif
struct rhc;
struct dds_qos;
struct ddsi_tkmap_instance;
struct ddsi_serdata;
struct ddsi_sertopic;
struct entity_common;
struct proxy_writer_info
{
nn_guid_t guid;
bool auto_dispose;
int32_t ownership_strength;
uint64_t iid;
};
struct ddsi_rhc_plugin
{
void (*rhc_free_fn) (struct rhc *rhc);
bool (*rhc_store_fn)
(struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info,
struct ddsi_serdata * __restrict sample, struct ddsi_tkmap_instance * __restrict tk);
void (*rhc_unregister_wr_fn)
(struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info);
void (*rhc_relinquish_ownership_fn)
(struct rhc * __restrict rhc, const uint64_t wr_iid);
void (*rhc_set_qos_fn) (struct rhc * rhc, const struct dds_qos * qos);
};
DDS_EXPORT void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct dds_qos *xqos);
#if defined (__cplusplus)
}
#endif
#endif

6
src/core/ddsi/include/dds/ddsi/q_config.h
View file

@ -20,7 +20,6 @@
#include "dds/ddsi/q_xqos.h"
#include "dds/ddsi/ddsi_tran.h"
#include "dds/ddsi/q_feature_check.h"
#include "dds/ddsi/ddsi_rhc_plugin.h"
#if defined (__cplusplus)
extern "C" {
@ -393,6 +392,8 @@ struct config
struct prune_deleted_ppant prune_deleted_ppant;
};
struct ddsi_sertopic;
struct entity_common;
struct ddsi_plugin
{
int (*init_fn) (void);
@ -402,9 +403,6 @@ struct ddsi_plugin
bool (*builtintopic_is_visible) (const nn_guid_t *guid, nn_vendorid_t vendorid);
struct ddsi_tkmap_instance * (*builtintopic_get_tkmap_entry) (const struct nn_guid *guid);
void (*builtintopic_write) (const struct entity_common *e, nn_wctime_t timestamp, bool alive);
/* Read cache */
struct ddsi_rhc_plugin rhc_plugin;
};
extern struct config DDS_EXPORT config;

82
src/core/ddsi/include/dds/ddsi/q_rhc.h Normal file
View file

@ -0,0 +1,82 @@
/*
* 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
*/
#ifndef Q_RHC_H
#define Q_RHC_H
#include <stddef.h>
#include <stdint.h>
#include <stdbool.h>
#include "dds/export.h"
#include "dds/ddsi/q_rtps.h"
#if defined (__cplusplus)
extern "C" {
#endif
struct dds_qos;
struct ddsi_tkmap_instance;
struct ddsi_serdata;
struct ddsi_sertopic;
struct entity_common;
struct proxy_writer_info
{
nn_guid_t guid;
bool auto_dispose;
int32_t ownership_strength;
uint64_t iid;
};
struct rhc;
typedef void (*rhc_free_t) (struct rhc *rhc);
typedef bool (*rhc_store_t) (struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info, struct ddsi_serdata * __restrict sample, struct ddsi_tkmap_instance * __restrict tk);
typedef void (*rhc_unregister_wr_t) (struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info);
typedef void (*rhc_relinquish_ownership_t) (struct rhc * __restrict rhc, const uint64_t wr_iid);
typedef void (*rhc_set_qos_t) (struct rhc *rhc, const struct dds_qos *qos);
struct rhc_ops {
rhc_store_t store;
rhc_unregister_wr_t unregister_wr;
rhc_relinquish_ownership_t relinquish_ownership;
rhc_set_qos_t set_qos;
rhc_free_t free;
};
struct rhc {
const struct rhc_ops *ops;
};
DDS_EXPORT inline bool 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) {
return rhc->ops->store (rhc, pwr_info, sample, tk);
}
DDS_EXPORT inline void rhc_unregister_wr (struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info) {
rhc->ops->unregister_wr (rhc, pwr_info);
}
DDS_EXPORT inline void rhc_relinquish_ownership (struct rhc * __restrict rhc, const uint64_t wr_iid) {
rhc->ops->relinquish_ownership (rhc, wr_iid);
}
DDS_EXPORT inline void rhc_set_qos (struct rhc *rhc, const struct dds_qos *qos) {
rhc->ops->set_qos (rhc, qos);
}
DDS_EXPORT inline void rhc_free (struct rhc *rhc) {
rhc->ops->free (rhc);
}
DDS_EXPORT void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct dds_qos *xqos);
#if defined (__cplusplus)
}
#endif
#endif /* Q_RHC_H */

11
src/core/ddsi/src/q_entity.c
View file

@ -42,6 +42,7 @@
#include "dds/ddsi/ddsi_mcgroup.h"
#include "dds/ddsi/q_receive.h"
#include "dds/ddsi/ddsi_udp.h" /* nn_mc4gen_address_t */
#include "dds/ddsi/q_rhc.h"
#include "dds/ddsi/sysdeps.h"
#include "dds__whc.h"
@ -1448,7 +1449,7 @@ static void reader_drop_connection (const struct nn_guid *rd_guid, const struct
{
struct proxy_writer_info pwr_info;
make_proxy_writer_info(&pwr_info, &pwr->e, pwr->c.xqos);
(ddsi_plugin.rhc_plugin.rhc_unregister_wr_fn) (rd->rhc, &pwr_info);
rhc_unregister_wr (rd->rhc, &pwr_info);
}
if (rd->status_cb)
{
@ -1483,7 +1484,7 @@ static void reader_drop_local_connection (const struct nn_guid *rd_guid, const s
/* FIXME: */
struct proxy_writer_info pwr_info;
make_proxy_writer_info(&pwr_info, &wr->e, wr->xqos);
(ddsi_plugin.rhc_plugin.rhc_unregister_wr_fn) (rd->rhc, &pwr_info);
rhc_unregister_wr (rd->rhc, &pwr_info);
}
if (rd->status_cb)
{
@ -1704,7 +1705,7 @@ static void writer_add_local_connection (struct writer *wr, struct reader *rd)
/* FIXME: whc has tk reference in its index nodes, which is what we really should be iterating over anyway, and so we don't really have to look them up anymore */
struct ddsi_tkmap_instance *tk = ddsi_tkmap_lookup_instance_ref(payload);
make_proxy_writer_info(&pwr_info, &wr->e, wr->xqos);
(void)(ddsi_plugin.rhc_plugin.rhc_store_fn) (rd->rhc, &pwr_info, payload, tk);
(void) rhc_store (rd->rhc, &pwr_info, payload, tk);
ddsi_tkmap_instance_unref(tk);
}
}
@ -3267,7 +3268,7 @@ static dds_return_t new_reader_guid
/* set rhc qos for reader */
if (rhc)
{
(ddsi_plugin.rhc_plugin.rhc_set_qos_fn) (rd->rhc, rd->xqos);
rhc_set_qos (rd->rhc, rd->xqos);
}
assert (rd->xqos->present & QP_LIVELINESS);
if (rd->xqos->liveliness.kind != DDS_LIVELINESS_AUTOMATIC || rd->xqos->liveliness.lease_duration != T_NEVER)
@ -3396,7 +3397,7 @@ static void gc_delete_reader (struct gcreq *gcreq)
#endif
if (rd->rhc)
{
(ddsi_plugin.rhc_plugin.rhc_free_fn) (rd->rhc);
rhc_free (rd->rhc);
}
if (rd->status_cb)
{

7
src/core/ddsi/src/q_receive.c
View file

@ -44,6 +44,7 @@
#include "dds/ddsi/q_entity.h"
#include "dds/ddsi/q_xmsg.h"
#include "dds/ddsi/q_receive.h"
#include "dds/ddsi/q_rhc.h"
#include "dds/ddsi/q_transmit.h"
#include "dds/ddsi/q_globals.h"
@ -1966,7 +1967,7 @@ static int deliver_user_data (const struct nn_rsample_info *sampleinfo, const st
for (uint32_t i = 0; rdary[i]; i++)
{
DDS_TRACE("reader "PGUIDFMT"\n", PGUID (rdary[i]->e.guid));
if (! (ddsi_plugin.rhc_plugin.rhc_store_fn) (rdary[i]->rhc, &pwr_info, payload, tk))
if (!rhc_store (rdary[i]->rhc, &pwr_info, payload, tk))
{
if (pwr_locked) ddsrt_mutex_unlock (&pwr->e.lock);
ddsrt_mutex_unlock (&pwr->rdary.rdary_lock);
@ -1997,7 +1998,7 @@ static int deliver_user_data (const struct nn_rsample_info *sampleinfo, const st
if ((rd = ephash_lookup_reader_guid (&m->rd_guid)) != NULL && m->in_sync == PRMSS_SYNC)
{
DDS_TRACE("reader-via-guid "PGUIDFMT"\n", PGUID (rd->e.guid));
(void) (ddsi_plugin.rhc_plugin.rhc_store_fn) (rd->rhc, &pwr_info, payload, tk);
(void) rhc_store (rd->rhc, &pwr_info, payload, tk);
}
}
if (!pwr_locked) ddsrt_mutex_unlock (&pwr->e.lock);
@ -2009,7 +2010,7 @@ static int deliver_user_data (const struct nn_rsample_info *sampleinfo, const st
{
struct reader *rd = ephash_lookup_reader_guid (rdguid);;
DDS_TRACE(" %"PRId64"=>"PGUIDFMT"%s\n", sampleinfo->seq, PGUID (*rdguid), rd ? "" : "?");
while (rd && ! (ddsi_plugin.rhc_plugin.rhc_store_fn) (rd->rhc, &pwr_info, payload, tk) && ephash_lookup_proxy_writer_guid (&pwr->e.guid))
while (rd && ! rhc_store (rd->rhc, &pwr_info, payload, tk) && ephash_lookup_proxy_writer_guid (&pwr->e.guid))
{
if (pwr_locked) ddsrt_mutex_unlock (&pwr->e.lock);
dds_sleepfor (DDS_MSECS (1));

12
src/core/ddsi/src/ddsi_rhc_plugin.c → src/core/ddsi/src/q_rhc.c
View file

@ -9,11 +9,17 @@
*
* SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
*/
#include "dds/ddsi/q_entity.h"
#include "dds/ddsi/q_rhc.h"
#include "dds/ddsi/q_xqos.h"
#include "dds/ddsi/ddsi_rhc_plugin.h"
#include "dds/ddsi/q_entity.h"
DDS_EXPORT void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct dds_qos *xqos)
extern inline void rhc_free (struct rhc *rhc);
extern inline bool 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);
extern inline void rhc_unregister_wr (struct rhc * __restrict rhc, const struct proxy_writer_info * __restrict pwr_info);
extern inline void rhc_relinquish_ownership (struct rhc * __restrict rhc, const uint64_t wr_iid);
extern inline void rhc_set_qos (struct rhc *rhc, const struct dds_qos *qos);
void make_proxy_writer_info(struct proxy_writer_info *pwr_info, const struct entity_common *e, const struct dds_qos *xqos)
{
pwr_info->guid = e->guid;
pwr_info->ownership_strength = xqos->ownership_strength.value;

29
src/core/xtests/rhc_torture/rhc_torture.c
View file

@ -31,6 +31,7 @@
#include "dds/ddsi/ddsi_serdata.h"
#include "dds__topic.h"
#include "dds__rhc.h"
#include "dds__rhc_default.h"
#include "dds/ddsi/ddsi_iid.h"
#include "RhcTypes.h"
@ -103,7 +104,7 @@ static struct ddsi_serdata *mkkeysample (int32_t keyval, unsigned statusinfo)
return sd;
}
static uint64_t store (struct rhc *rhc, struct proxy_writer *wr, struct ddsi_serdata *sd, bool print)
static uint64_t store (struct dds_rhc *rhc, struct proxy_writer *wr, struct ddsi_serdata *sd, bool print)
{
/* beware: unrefs sd */
struct ddsi_tkmap_instance *tk;
@ -161,9 +162,9 @@ static void fwr (struct proxy_writer *wr)
free (wr);
}
static struct rhc *mkrhc (dds_reader *rd, dds_history_kind_t hk, int32_t hdepth, dds_destination_order_kind_t dok)
static struct dds_rhc *mkrhc (dds_reader *rd, dds_history_kind_t hk, int32_t hdepth, dds_destination_order_kind_t dok)
{
struct rhc *rhc;
struct dds_rhc *rhc;
dds_qos_t rqos;
nn_xqos_init_empty (&rqos);
rqos.present |= QP_HISTORY | QP_DESTINATION_ORDER;
@ -172,13 +173,13 @@ static struct rhc *mkrhc (dds_reader *rd, dds_history_kind_t hk, int32_t hdepth,
rqos.destination_order.kind = dok;
nn_xqos_mergein_missing (&rqos, &gv.default_xqos_rd, ~(uint64_t)0);
thread_state_awake (lookup_thread_state ());
rhc = dds_rhc_new (rd, mdtopic);
rhc = dds_rhc_default_new (rd, mdtopic);
dds_rhc_set_qos(rhc, &rqos);
thread_state_asleep (lookup_thread_state ());
return rhc;
}
static void frhc (struct rhc *rhc)
static void frhc (struct dds_rhc *rhc)
{
thread_state_awake (lookup_thread_state ());
dds_rhc_free (rhc);
@ -281,7 +282,7 @@ static void print_seq (int n, const dds_sample_info_t *iseq, const RhcTypes_T *m
}
}
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])
static void rdtkcond (struct dds_rhc *rhc, dds_readcond *cond, const struct check *chk, bool print, int max, const char *opname, int (*op) (struct dds_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;
@ -379,12 +380,12 @@ static void rdtkcond (struct rhc *rhc, dds_readcond *cond, const struct check *c
}
}
static void rdall (struct rhc *rhc, const struct check *chk, bool print, uint32_t states_seen[STATIC_ARRAY_DIM 2*2*3][2])
static void rdall (struct dds_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])
static void tkall (struct dds_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);
}
@ -440,7 +441,7 @@ static void print_condmask (char *buf, size_t bufsz, const dds_readcond *cond)
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])
static void rdcond (struct dds_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;
@ -449,7 +450,7 @@ static void rdcond (struct rhc *rhc, dds_readcond *cond, const struct check *chk
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])
static void tkcond (struct dds_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;
@ -539,14 +540,14 @@ static void test_conditions (dds_entity_t pp, dds_entity_t tp, const int count,
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])];
struct dds_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;
rhc[i] = rdp->m_rhc;
dds_entity_unlock (x);
}
struct proxy_writer *wr[] = { mkwr (0), mkwr (1), mkwr (1) };
@ -848,7 +849,7 @@ int main (int argc, char **argv)
{
if (print)
printf ("************* 0 *************\n");
struct rhc *rhc = mkrhc (NULL, DDS_HISTORY_KEEP_LAST, 1, DDS_DESTINATIONORDER_BY_SOURCE_TIMESTAMP);
struct dds_rhc *rhc = mkrhc (NULL, DDS_HISTORY_KEEP_LAST, 1, DDS_DESTINATIONORDER_BY_SOURCE_TIMESTAMP);
struct proxy_writer *wr0 = mkwr (1);
uint64_t iid0, iid1, iid_t;
iid0 = store (rhc, wr0, mksample (0, 0), print);
@ -894,7 +895,7 @@ int main (int argc, char **argv)
{
if (print)
printf ("************* 1 *************\n");
struct rhc *rhc = mkrhc (NULL, DDS_HISTORY_KEEP_LAST, 4, DDS_DESTINATIONORDER_BY_SOURCE_TIMESTAMP);
struct dds_rhc *rhc = mkrhc (NULL, DDS_HISTORY_KEEP_LAST, 4, DDS_DESTINATIONORDER_BY_SOURCE_TIMESTAMP);
struct proxy_writer *wr[] = { mkwr (0), mkwr (0), mkwr (0) };
uint64_t iid0, iid_t;
int nregs = 3, isreg[] = { 1, 1, 1 };