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

Give all entities an instance handle (#43)

Browse source 
Add the instance handle to the DDSC entity type, initialize it properly
for all types, and remove the per-type handling of
dds_get_instance_handle.  Those entities that have a DDSI variant take
the instance handle from DDSI (which plays tricks to get the instance
handles of the entities matching the built-in topics).  For those that
do not have a DDSI variant, just generate a unique identifier using the
same generate that DDSI uses.

Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson 2019-06-08 19:23:53 +02:00 • committed by eboasson
parent 9c2f3bdf2b
commit 28317ba49e
16 changed files with 70 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -103,41 +103,45 @@ dds_domain;
struct dds_entity;
typedef struct dds_entity_deriver {
/* Close can be used to terminate (blocking) actions on a entity before actually deleting it. */
dds_return_t (*close)(struct dds_entity *e) ddsrt_nonnull_all;
/* Delete is used to actually free the entity. */
dds_return_t (*delete)(struct dds_entity *e) ddsrt_nonnull_all;
dds_return_t (*set_qos)(struct dds_entity *e, const dds_qos_t *qos, bool enabled) ddsrt_nonnull_all;
dds_return_t (*validate_status)(uint32_t mask);
dds_return_t (*get_instance_hdl)(struct dds_entity *e, dds_instance_handle_t *i) ddsrt_nonnull_all;
}
dds_entity_deriver;
/* Close can be used to terminate (blocking) actions on a entity before actually deleting it. */
dds_return_t (*close)(struct dds_entity *e) ddsrt_nonnull_all;
/* Delete is used to actually free the entity. */
dds_return_t (*delete)(struct dds_entity *e) ddsrt_nonnull_all;
dds_return_t (*set_qos)(struct dds_entity *e, const dds_qos_t *qos, bool enabled) ddsrt_nonnull_all;
dds_return_t (*validate_status)(uint32_t mask);
} dds_entity_deriver;
typedef void (*dds_entity_callback)(dds_entity_t observer, dds_entity_t observed, uint32_t status);
typedef struct dds_entity_observer
{
dds_entity_callback m_cb;
dds_entity_t m_observer;
struct dds_entity_observer *m_next;
}
dds_entity_observer;
typedef struct dds_entity_observer {
dds_entity_callback m_cb;
dds_entity_t m_observer;
struct dds_entity_observer *m_next;
} dds_entity_observer;
typedef struct dds_entity
{
struct dds_handle_link m_hdllink;
dds_entity_kind_t m_kind;
dds_entity_deriver m_deriver;
uint32_t m_refc;
struct dds_entity * m_next;
struct dds_entity * m_parent;
struct dds_entity * m_children;
struct dds_entity * m_participant;
struct dds_domain * m_domain;
dds_qos_t * m_qos;
dds_domainid_t m_domainid;
nn_guid_t m_guid;
uint32_t m_flags;
typedef struct dds_entity {
struct dds_handle_link m_hdllink; /* handle is constant, cnt_flags private to dds_handle.c */
dds_entity_kind_t m_kind; /* constant */
dds_entity_deriver m_deriver; /* constant; FIXME: no point in having function pointers embedded */
uint32_t m_refc; /* [m_mutex] */
struct dds_entity *m_next; /* [m_mutex] */
struct dds_entity *m_parent; /* constant */
struct dds_entity *m_children; /* [m_mutex] */
struct dds_entity *m_participant; /* constant */
struct dds_domain *m_domain; /* constant */
dds_qos_t *m_qos; /* [m_mutex] */
dds_domainid_t m_domainid; /* constant; FIXME: why? hardly ever used, m_domain should give that info, too */
nn_guid_t m_guid; /* ~ constant: FIXME: set during creation, but possibly after becoming visible */
dds_instance_handle_t m_iid; /* ~ constant: FIXME: like GUID */
uint32_t m_flags; /* [m_mutex] */
/* Allowed:
- locking parent->...->m_mutex while holding m_mutex
- locking topic::m_mutex while holding {publisher,subscriber}::m_mutex
(no hierarchical relationship there)
- locking topic::m_mutex while holding {reader,writer}::m_mutex
- locking observers_lock while holding m_mutex
*/
ddsrt_mutex_t m_mutex;
ddsrt_cond_t m_cond;

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

@ -889,14 +889,10 @@ dds_return_t dds_get_instance_handle (dds_entity_t entity, dds_instance_handle_t
if (ihdl == NULL)
return DDS_RETCODE_BAD_PARAMETER;
if ((ret = dds_entity_lock (entity, DDS_KIND_DONTCARE, &e)) != DDS_RETCODE_OK)
if ((ret = dds_entity_claim (entity, &e)) != DDS_RETCODE_OK)
return ret;
if (e->m_deriver.get_instance_hdl)
ret = e->m_deriver.get_instance_hdl (e, ihdl);
else
ret = DDS_RETCODE_ILLEGAL_OPERATION;
dds_entity_unlock(e);
*ihdl = e->m_iid;
dds_entity_release(e);
return ret;
}

4
src/core/ddsc/src/dds_guardcond.c
View file

@ -14,6 +14,7 @@
#include "dds__reader.h"
#include "dds__guardcond.h"
#include "dds__participant.h"
#include "dds/ddsi/ddsi_iid.h"
#include "dds/ddsi/q_ephash.h"
#include "dds/ddsi/q_entity.h"
#include "dds/ddsi/q_thread.h"
@ -29,8 +30,9 @@ dds_entity_t dds_create_guardcondition (dds_entity_t participant)
return rc;
else
{
dds_guardcond * gcond = dds_alloc (sizeof (*gcond));
dds_guardcond *gcond = dds_alloc (sizeof (*gcond));
dds_entity_t hdl = dds_entity_init (&gcond->m_entity, &pp->m_entity, DDS_KIND_COND_GUARD, NULL, NULL, 0);
gcond->m_entity.m_iid = ddsi_iid_gen ();
dds_participant_unlock (pp);
return hdl;
}

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

@ -64,14 +64,6 @@ static dds_return_t dds_participant_delete (dds_entity *e)
return DDS_RETCODE_OK;
}
static dds_return_t dds_participant_instance_hdl (dds_entity *e, dds_instance_handle_t *i) ddsrt_nonnull_all;
static dds_return_t dds_participant_instance_hdl (dds_entity *e, dds_instance_handle_t *i)
{
*i = participant_instance_id (&e->m_guid);
return DDS_RETCODE_OK;
}
static dds_return_t dds_participant_qos_set (dds_entity *e, const dds_qos_t *qos, bool enabled)
{
/* note: e->m_qos is still the old one to allow for failure here */
@ -134,11 +126,11 @@ dds_entity_t dds_create_participant (const dds_domainid_t domain, const dds_qos_
goto err_entity_init;
pp->m_entity.m_guid = guid;
pp->m_entity.m_iid = get_entity_instance_id (&guid);
pp->m_entity.m_domain = dds_domain_create (dds_domain_default ());
pp->m_entity.m_domainid = dds_domain_default ();
pp->m_entity.m_deriver.delete = dds_participant_delete;
pp->m_entity.m_deriver.set_qos = dds_participant_qos_set;
pp->m_entity.m_deriver.get_instance_hdl = dds_participant_instance_hdl;
pp->m_entity.m_deriver.validate_status = dds_participant_status_validate;
pp->m_builtin_subscriber = 0;

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

@ -24,15 +24,6 @@ DECL_ENTITY_LOCK_UNLOCK (extern inline, dds_publisher)
#define DDS_PUBLISHER_STATUS_MASK (0u)
static dds_return_t dds_publisher_instance_hdl (dds_entity *e, dds_instance_handle_t *i) ddsrt_nonnull_all;
static dds_return_t dds_publisher_instance_hdl (dds_entity *e, dds_instance_handle_t *i)
{
/* FIXME: Get/generate proper handle. */
(void) e; (void) i;
return DDS_RETCODE_UNSUPPORTED;
}
static dds_return_t dds_publisher_qos_set (dds_entity *e, const dds_qos_t *qos, bool enabled)
{
/* note: e->m_qos is still the old one to allow for failure here */
@ -70,8 +61,8 @@ dds_entity_t dds_create_publisher (dds_entity_t participant, const dds_qos_t *qo
}
pub = dds_alloc (sizeof (*pub));
hdl = dds_entity_init (&pub->m_entity, &par->m_entity, DDS_KIND_PUBLISHER, new_qos, listener, DDS_PUBLISHER_STATUS_MASK);
pub->m_entity.m_iid = ddsi_iid_gen ();
pub->m_entity.m_deriver.set_qos = dds_publisher_qos_set;
pub->m_entity.m_deriver.get_instance_hdl = dds_publisher_instance_hdl;
pub->m_entity.m_deriver.validate_status = dds_publisher_status_validate;
dds_participant_unlock (par);
return hdl;

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

@ -14,6 +14,7 @@
#include "dds__readcond.h"
#include "dds__rhc.h"
#include "dds__entity.h"
#include "dds/ddsi/ddsi_iid.h"
#include "dds/ddsi/q_ephash.h"
#include "dds/ddsi/q_entity.h"
#include "dds/ddsi/q_thread.h"
@ -31,6 +32,7 @@ dds_readcond *dds_create_readcond (dds_reader *rd, dds_entity_kind_t kind, uint3
dds_readcond *cond = dds_alloc (sizeof (*cond));
assert ((kind == DDS_KIND_COND_READ && filter == 0) || (kind == DDS_KIND_COND_QUERY && filter != 0));
(void) dds_entity_init (&cond->m_entity, &rd->m_entity, kind, NULL, NULL, 0);
cond->m_entity.m_iid = ddsi_iid_gen ();
cond->m_entity.m_deriver.delete = dds_readcond_delete;
cond->m_rhc = rd->m_rd->rhc;
cond->m_sample_states = mask & DDS_ANY_SAMPLE_STATE;

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

@ -39,14 +39,6 @@ DECL_ENTITY_LOCK_UNLOCK (extern inline, dds_reader)
DDS_SAMPLE_LOST_STATUS |\
DDS_SUBSCRIPTION_MATCHED_STATUS)
static dds_return_t dds_reader_instance_hdl (dds_entity *e, dds_instance_handle_t *i) ddsrt_nonnull_all;
static dds_return_t dds_reader_instance_hdl (dds_entity *e, dds_instance_handle_t *i)
{
*i = reader_instance_id (&e->m_guid);
return DDS_RETCODE_OK;
}
static dds_return_t dds_reader_close (dds_entity *e) ddsrt_nonnull_all;
static dds_return_t dds_reader_close (dds_entity *e)
@ -392,7 +384,6 @@ dds_entity_t dds_create_reader (dds_entity_t participant_or_subscriber, dds_enti
rd->m_entity.m_deriver.delete = dds_reader_delete;
rd->m_entity.m_deriver.set_qos = dds_reader_qos_set;
rd->m_entity.m_deriver.validate_status = dds_reader_status_validate;
rd->m_entity.m_deriver.get_instance_hdl = dds_reader_instance_hdl;
/* Extra claim of this reader to make sure that the delete waits until DDSI
has deleted its reader as well. This can be known through the callback. */
@ -407,6 +398,7 @@ dds_entity_t dds_create_reader (dds_entity_t participant_or_subscriber, dds_enti
ddsrt_mutex_lock (&tp->m_entity.m_mutex);
assert (ret == DDS_RETCODE_OK); /* FIXME: can be out-of-resources at the very least */
thread_state_asleep (lookup_thread_state ());
rd->m_entity.m_iid = get_entity_instance_id (&rd->m_entity.m_guid);
/* For persistent data register reader with durability */
if (dds_global.m_dur_reader && (rd->m_entity.m_qos->durability.kind > DDS_DURABILITY_TRANSIENT_LOCAL)) {

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

@ -23,15 +23,6 @@ DECL_ENTITY_LOCK_UNLOCK (extern inline, dds_subscriber)
#define DDS_SUBSCRIBER_STATUS_MASK \
(DDS_DATA_ON_READERS_STATUS)
static dds_return_t dds_subscriber_instance_hdl (dds_entity *e, dds_instance_handle_t *i) ddsrt_nonnull_all;
static dds_return_t dds_subscriber_instance_hdl (dds_entity *e, dds_instance_handle_t *i)
{
(void) e; (void) i;
/* FIXME: Get/generate proper handle. */
return DDS_RETCODE_UNSUPPORTED;
}
static dds_return_t dds_subscriber_qos_set (dds_entity *e, const dds_qos_t *qos, bool enabled)
{
/* note: e->m_qos is still the old one to allow for failure here */
@ -64,9 +55,9 @@ dds_entity_t dds__create_subscriber_l (dds_participant *participant, const dds_q
sub = dds_alloc (sizeof (*sub));
subscriber = dds_entity_init (&sub->m_entity, &participant->m_entity, DDS_KIND_SUBSCRIBER, new_qos, listener, DDS_SUBSCRIBER_STATUS_MASK);
sub->m_entity.m_iid = ddsi_iid_gen ();
sub->m_entity.m_deriver.set_qos = dds_subscriber_qos_set;
sub->m_entity.m_deriver.validate_status = dds_subscriber_status_validate;
sub->m_entity.m_deriver.get_instance_hdl = dds_subscriber_instance_hdl;
return subscriber;
}

1
src/core/ddsc/src/dds_topic.c
View file

@ -271,6 +271,7 @@ dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_s
/* Create topic */
top = dds_alloc (sizeof (*top));
hdl = dds_entity_init (&top->m_entity, &par->m_entity, DDS_KIND_TOPIC, new_qos, listener, DDS_TOPIC_STATUS_MASK);
top->m_entity.m_iid = ddsi_iid_gen ();
top->m_entity.m_deriver.delete = dds_topic_delete;
top->m_entity.m_deriver.set_qos = dds_topic_qos_set;
top->m_entity.m_deriver.validate_status = dds_topic_status_validate;

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

@ -18,6 +18,7 @@
#include "dds__querycond.h"
#include "dds__readcond.h"
#include "dds__rhc.h"
#include "dds/ddsi/ddsi_iid.h"
DEFINE_ENTITY_LOCK_UNLOCK (static, dds_waitset, DDS_KIND_WAITSET)
@ -153,6 +154,7 @@ dds_entity_t dds_create_waitset (dds_entity_t participant)
dds_waitset *waitset = dds_alloc (sizeof (*waitset));
hdl = dds_entity_init (&waitset->m_entity, &par->m_entity, DDS_KIND_WAITSET, NULL, NULL, 0);
waitset->m_entity.m_iid = ddsi_iid_gen ();
waitset->m_entity.m_deriver.close = dds_waitset_close;
waitset->observed = NULL;
waitset->triggered = NULL;

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

@ -36,14 +36,6 @@ DECL_ENTITY_LOCK_UNLOCK (extern inline, dds_writer)
DDS_OFFERED_INCOMPATIBLE_QOS_STATUS |\
DDS_PUBLICATION_MATCHED_STATUS)
static dds_return_t dds_writer_instance_hdl (dds_entity *e, dds_instance_handle_t *i) ddsrt_nonnull_all;
static dds_return_t dds_writer_instance_hdl (dds_entity *e, dds_instance_handle_t *i)
{
*i = writer_instance_id(&e->m_guid);
return DDS_RETCODE_OK;
}
static dds_return_t dds_writer_status_validate (uint32_t mask)
{
return (mask & ~DDS_WRITER_STATUS_MASK) ? DDS_RETCODE_BAD_PARAMETER : DDS_RETCODE_OK;
@ -305,7 +297,6 @@ dds_entity_t dds_create_writer (dds_entity_t participant_or_publisher, dds_entit
wr->m_entity.m_deriver.delete = dds_writer_delete;
wr->m_entity.m_deriver.set_qos = dds_writer_qos_set;
wr->m_entity.m_deriver.validate_status = dds_writer_status_validate;
wr->m_entity.m_deriver.get_instance_hdl = dds_writer_instance_hdl;
wr->m_whc = make_whc (wqos);
/* Extra claim of this writer to make sure that the delete waits until DDSI
@ -321,6 +312,7 @@ dds_entity_t dds_create_writer (dds_entity_t participant_or_publisher, dds_entit
ddsrt_mutex_lock (&tp->m_entity.m_mutex);
assert(rc == DDS_RETCODE_OK);
thread_state_asleep (lookup_thread_state ());
wr->m_entity.m_iid = get_entity_instance_id (&wr->m_entity.m_guid);
dds_topic_unlock (tp);
dds_publisher_unlock (pub);
return writer;

11
src/core/ddsc/tests/unsupported.c
View file

@ -121,16 +121,19 @@ CU_Test(ddsc_unsupported, dds_get_instance_handle, .init = setup, .fini = teardo
dds_return_t result;
dds_instance_handle_t ih;
static struct index_result pars[] = {
{TOP, DDS_RETCODE_ILLEGAL_OPERATION}, /* TODO: Shouldn't this be either supported or unsupported? */
{PUB, DDS_RETCODE_UNSUPPORTED},
{SUB, DDS_RETCODE_UNSUPPORTED},
{RCD, DDS_RETCODE_ILLEGAL_OPERATION},
{TOP, DDS_RETCODE_OK},
{PUB, DDS_RETCODE_OK},
{SUB, DDS_RETCODE_OK},
{RCD, DDS_RETCODE_OK},
{BAD, DDS_RETCODE_BAD_PARAMETER}
};
for (size_t i=0; i < sizeof (pars) / sizeof (pars[0]);i++) {
result = dds_get_instance_handle(e[pars[i].index], &ih);
CU_ASSERT_EQUAL(result, pars[i].exp_res);
if (pars[i].exp_res == DDS_RETCODE_OK) {
CU_ASSERT(ih > 0);
}
}
}

5
src/core/ddsi/include/dds/ddsi/q_entity.h
View file

@ -529,6 +529,7 @@ dds_return_t new_participant (struct nn_guid *ppguid, unsigned flags, const stru
*/
dds_return_t delete_participant (const struct nn_guid *ppguid);
void update_participant_plist (struct participant *pp, const struct nn_plist *plist);
uint64_t get_entity_instance_id (const struct nn_guid *guid);
/* To obtain the builtin writer to be used for publishing SPDP, SEDP,
PMD stuff for PP and its endpoints, given the entityid. If PP has
@ -559,7 +560,6 @@ dds_return_t delete_writer_nolinger (const struct nn_guid *guid);
dds_return_t delete_writer_nolinger_locked (struct writer *wr);
dds_return_t delete_reader (const struct nn_guid *guid);
uint64_t reader_instance_id (const struct nn_guid *guid);
struct local_orphan_writer {
struct writer wr;
@ -593,7 +593,6 @@ void delete_local_orphan_writer (struct local_orphan_writer *wr);
void new_proxy_participant (const struct nn_guid *guid, unsigned bes, unsigned prismtech_bes, const struct nn_guid *privileged_pp_guid, struct addrset *as_default, struct addrset *as_meta, const struct nn_plist *plist, dds_duration_t tlease_dur, nn_vendorid_t vendor, unsigned custom_flags, nn_wctime_t timestamp, seqno_t seq);
int delete_proxy_participant_by_guid (const struct nn_guid * guid, nn_wctime_t timestamp, int isimplicit);
uint64_t participant_instance_id (const struct nn_guid *guid);
enum update_proxy_participant_source {
UPD_PROXYPP_SPDP,
@ -629,8 +628,6 @@ void update_proxy_writer (struct proxy_writer *pwr, struct addrset *as, const st
int new_proxy_group (const struct nn_guid *guid, const char *name, const struct dds_qos *xqos, nn_wctime_t timestamp);
void delete_proxy_group (const struct nn_guid *guid, nn_wctime_t timestamp, int isimplicit);
uint64_t writer_instance_id (const struct nn_guid *guid);
/* Call this to empty all address sets of all writers to stop all outgoing traffic, or to
rebuild them all (which only makes sense after previously having emptied them all). */
void rebuild_or_clear_writer_addrsets(int rebuild);

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

@ -3124,20 +3124,6 @@ dds_return_t delete_writer (const struct nn_guid *guid)
return 0;
}
uint64_t writer_instance_id (const struct nn_guid *guid)
{
struct entity_common *e;
e = (struct entity_common*)ephash_lookup_writer_guid(guid);
if (e) {
return e->iid;
}
e = (struct entity_common*)ephash_lookup_proxy_writer_guid(guid);
if (e) {
return e->iid;
}
return 0;
}
/* READER ----------------------------------------------------------- */
#ifdef DDSI_INCLUDE_NETWORK_PARTITIONS
@ -3464,20 +3450,6 @@ dds_return_t delete_reader (const struct nn_guid *guid)
return 0;
}
uint64_t reader_instance_id (const struct nn_guid *guid)
{
struct entity_common *e;
e = (struct entity_common*)ephash_lookup_reader_guid(guid);
if (e) {
return e->iid;
}
e = (struct entity_common*)ephash_lookup_proxy_reader_guid(guid);
if (e) {
return e->iid;
}
return 0;
}
void update_reader_qos (struct reader *rd, const dds_qos_t *xqos)
{
ddsrt_mutex_lock (&rd->e.lock);
@ -3967,18 +3939,16 @@ int delete_proxy_participant_by_guid (const struct nn_guid * guid, nn_wctime_t t
return 0;
}
uint64_t participant_instance_id (const struct nn_guid *guid)
uint64_t get_entity_instance_id (const struct nn_guid *guid)
{
struct entity_common *e;
e = (struct entity_common*)ephash_lookup_participant_guid(guid);
if (e) {
return e->iid;
}
e = (struct entity_common*)ephash_lookup_proxy_participant_guid(guid);
if (e) {
return e->iid;
}
return 0;
struct thread_state1 *ts1 = lookup_thread_state ();
struct entity_common *e;
uint64_t iid = 0;
thread_state_awake (ts1);
if ((e = ephash_lookup_guid_untyped (guid)) != NULL)
iid = e->iid;
thread_state_asleep (ts1);
return iid;
}
/* PROXY-GROUP --------------------------------------------------- */

1
src/core/ddsi/src/q_ephash.c
View file

@ -121,6 +121,7 @@ void *ephash_lookup_guid_untyped (const struct nn_guid *guid)
/* FIXME: could (now) require guid to be first in entity_common; entity_common already is first in entity */
struct entity_common e;
e.guid = *guid;
assert (thread_is_awake ());
return ddsrt_chh_lookup (gv.guid_hash->hash, &e);
}

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

@ -542,9 +542,9 @@ void xeventq_free (struct xeventq *evq)
handle_nontimed_xevent (getnext_from_non_timed_xmit_list (evq), xp);
}
ddsrt_mutex_unlock (&evq->lock);
thread_state_asleep (lookup_thread_state ());
nn_xpack_send (xp, false);
nn_xpack_free (xp);
thread_state_asleep (lookup_thread_state ());
}
assert (ddsrt_avl_is_empty (&evq->msg_xevents));