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

rework built-in topics again

Browse source 
Move details of built-in topics out of the DDSI core (so the only hooks
remain).  For this, rtps_term had to be split, so now it is "stop"
followed by "fini".

Add a notion of local writers that are not bound to a participant ("local
orphans"), so that the local built-in topic writers can be created during
initialization.  This eliminates the "builtin" participant.  This
uncovered in inconsistency in the unit tests: on the one hand, a newly
created participant is expected to have no child entities; on the other
hand, the built-in topics were expected to be returned by find_topic ...
This inconsistency has been resolved by creating them lazily and
accepting that find_topic can't return them until they have been
created.  Special code was in place in dds_create_reader anyway, so it
is not expected to have any real consequence for applications.

Use a special WHC implementation that regenerates the data on the fly
using the internal discovery tables of DDSI, so that the samples are only
stored by readers.  This eliminates the memory overhead of that existed
previously when the WHC of the writers stored the data.

No longer return topic name and type name in the built-in topics, they
have been extracted already and are not accessible through the normal
interface but do cause problems when comparing QoS.

Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson 2019-01-06 13:10:24 +01:00
parent d6dcb0558d
commit 66076817e1
26 changed files with 876 additions and 919 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -20,10 +20,8 @@ PREPEND(srcs_ddsi "${CMAKE_CURRENT_LIST_DIR}/src"
ddsi_mcgroup.c
ddsi_serdata.c
ddsi_serdata_default.c
ddsi_serdata_builtin.c
ddsi_sertopic.c
ddsi_sertopic_default.c
ddsi_sertopic_builtin.c
ddsi_rhc_plugin.c
ddsi_iid.c
ddsi_tkmap.c
@ -76,7 +74,6 @@ PREPEND(hdrs_private_ddsi "${CMAKE_CURRENT_LIST_DIR}/include/ddsi"
ddsi_serdata.h
ddsi_sertopic.h
ddsi_serdata_default.h
ddsi_serdata_builtin.h
ddsi_rhc_plugin.h
ddsi_iid.h
ddsi_tkmap.h

43
src/core/ddsi/include/ddsi/ddsi_serdata_builtin.h
View file

@ -1,43 +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_SERDATA_BUILTIN_H
#define DDSI_SERDATA_BUILTIN_H
#include "os/os.h"
#include "util/ut_avl.h"
#include "ddsi/ddsi_serdata.h"
#include "ddsi/ddsi_sertopic.h"
#include "ddsi/q_xqos.h"
struct ddsi_serdata_builtin {
struct ddsi_serdata c;
nn_guid_t key;
nn_xqos_t xqos;
};
enum ddsi_sertopic_builtin_type {
DSBT_PARTICIPANT,
DSBT_READER,
DSBT_WRITER
};
struct ddsi_sertopic_builtin {
struct ddsi_sertopic c;
enum ddsi_sertopic_builtin_type type;
};
extern const struct ddsi_sertopic_ops ddsi_sertopic_ops_builtin;
extern const struct ddsi_serdata_ops ddsi_serdata_ops_builtin;
struct ddsi_sertopic *new_sertopic_builtin (enum ddsi_sertopic_builtin_type type, const char *name, const char *typename);
#endif

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

@ -22,7 +22,6 @@
#include "ddsi/q_xqos.h"
#include "ddsi/ddsi_tran.h"
#include "ddsi/q_feature_check.h"
#include "ddsi/ddsi_serdata_builtin.h"
#include "ddsi/ddsi_rhc_plugin.h"
#if defined (__cplusplus)
@ -412,7 +411,10 @@ struct ddsi_plugin
{
int (*init_fn) (void);
void (*fini_fn) (void);
void (*builtin_write) (enum ddsi_sertopic_builtin_type type, const nn_guid_t *guid, nn_wctime_t timestamp, bool alive);
bool (*builtintopic_is_visible) (nn_entityid_t entityid, bool onlylocal, 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;

16
src/core/ddsi/include/ddsi/q_entity.h
View file

@ -127,12 +127,15 @@ struct pwr_rd_match {
struct nn_rsample_info;
struct nn_rdata;
struct ddsi_tkmap_instance;
struct entity_common {
enum entity_kind kind;
nn_guid_t guid;
nn_wctime_t tupdate; /* timestamp of last update */
char *name;
uint64_t iid;
struct ddsi_tkmap_instance *tk;
os_mutex lock;
bool onlylocal;
};
@ -390,8 +393,11 @@ int is_deleted_participant_guid (const struct nn_guid *guid, unsigned for_what);
nn_entityid_t to_entityid (unsigned u);
int is_builtin_entityid (nn_entityid_t id, nn_vendorid_t vendorid);
int is_builtin_endpoint (nn_entityid_t id, nn_vendorid_t vendorid);
bool is_local_orphan_endpoint (const struct entity_common *e);
int is_writer_entityid (nn_entityid_t id);
int is_reader_entityid (nn_entityid_t id);
nn_vendorid_t get_entity_vendorid (const struct entity_common *e);
int pp_allocate_entityid (nn_entityid_t *id, unsigned kind, struct participant *pp);
void pp_release_entityid(struct participant *pp, nn_entityid_t id);
@ -473,9 +479,9 @@ struct writer *get_builtin_writer (const struct participant *pp, unsigned entity
GUID "ppguid". May return NULL if participant unknown or
writer/reader already known. */
struct writer * new_writer (struct nn_guid *wrguid, const struct nn_guid *group_guid, const struct nn_guid *ppguid, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc * whc, status_cb_t status_cb, void * status_cb_arg);
struct writer *new_writer (struct nn_guid *wrguid, const struct nn_guid *group_guid, const struct nn_guid *ppguid, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc * whc, status_cb_t status_cb, void *status_cb_arg);
struct reader * new_reader (struct nn_guid *rdguid, const struct nn_guid *group_guid, const struct nn_guid *ppguid, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct rhc * rhc, status_cb_t status_cb, void * status_cb_arg);
struct reader *new_reader (struct nn_guid *rdguid, const struct nn_guid *group_guid, const struct nn_guid *ppguid, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct rhc * rhc, status_cb_t status_cb, void *status_cb_arg);
struct whc_node;
struct whc_state;
@ -492,6 +498,12 @@ int delete_writer_nolinger_locked (struct writer *wr);
int delete_reader (const struct nn_guid *guid);
uint64_t reader_instance_id (const struct nn_guid *guid);
struct local_orphan_writer {
struct writer wr;
};
struct local_orphan_writer *new_local_orphan_writer (nn_entityid_t entityid, struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc *whc);
void delete_local_orphan_writer (struct local_orphan_writer *wr);
/* To create or delete a new proxy participant: "guid" MUST have the
pre-defined participant entity id. Unlike delete_participant(),
deleting a proxy participant will automatically delete all its

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

@ -43,6 +43,7 @@ struct gcreq {
};
struct gcreq_queue *gcreq_queue_new (void);
void gcreq_queue_drain (struct gcreq_queue *q);
void gcreq_queue_free (struct gcreq_queue *q);
struct gcreq *gcreq_new (struct gcreq_queue *gcreq_queue, gcreq_cb_t cb);

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

@ -281,11 +281,6 @@ struct q_globals {
struct ddsi_sertopic *plist_topic; /* used for all discovery data */
struct ddsi_sertopic *rawcdr_topic; /* used for participant message data */
/* Sertopics for built-in topics -- FIXME: these really have little to do with topics, but everything with topic types, in other words, they are the type supports in DDS ... so a bit of refactoring is required */
struct ddsi_sertopic *builtin_participant_topic;
struct ddsi_sertopic *builtin_reader_topic;
struct ddsi_sertopic *builtin_writer_topic;
/* Network ID needed by v_groupWrite -- FIXME: might as well pass it
to the receive thread instead of making it global (and that would
remove the need to include kernelModule.h) */

3
src/core/ddsi/include/ddsi/q_rtps.h
View file

@ -80,7 +80,8 @@ int rtps_config_prep (struct cfgst *cfgst);
int rtps_config_open (void);
int rtps_init (void);
void ddsi_plugin_init (void);
void rtps_term (void);
void rtps_stop (void);
void rtps_fini (void);
#if defined (__cplusplus)
}

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

@ -42,7 +42,7 @@ struct whc_state {
an iter on the stack without specifying an implementation. If future changes or
implementations require more, these can be adjusted. An implementation should check
things fit at compile time. */
#define WHC_SAMPLE_ITER_SIZE (1*sizeof(void *))
#define WHC_SAMPLE_ITER_SIZE (7 * sizeof(void *))
struct whc_sample_iter_base {
struct whc *whc;
};

295
src/core/ddsi/src/ddsi_serdata_builtin.c
View file

@ -1,295 +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
*/
#include <stddef.h>
#include <ctype.h>
#include <assert.h>
#include <string.h>
#include "os/os.h"
#include "ddsi/q_md5.h"
#include "ddsi/q_bswap.h"
#include "ddsi/q_config.h"
#include "ddsi/q_freelist.h"
#include <assert.h>
#include <string.h>
#include "os/os.h"
#include "dds__key.h"
#include "ddsi/ddsi_tkmap.h"
#include "dds__stream.h"
#include "ddsi/q_entity.h"
#include "ddsi/ddsi_serdata_builtin.h"
//#include "dds.h" /* FIXME: need the sample types of the built-in topics */
static const uint64_t unihashconsts[] = {
UINT64_C (16292676669999574021),
UINT64_C (10242350189706880077),
UINT64_C (12844332200329132887),
UINT64_C (16728792139623414127)
};
static uint32_t hash_guid (const nn_guid_t *g)
{
return
(uint32_t) (((((uint32_t) g->prefix.u[0] + unihashconsts[0]) *
((uint32_t) g->prefix.u[1] + unihashconsts[1])) +
(((uint32_t) g->prefix.u[2] + unihashconsts[2]) *
((uint32_t) g->entityid.u + unihashconsts[3])))
>> 32);
}
static struct ddsi_serdata *fix_serdata_builtin(struct ddsi_serdata_builtin *d, uint32_t basehash)
{
d->c.hash = hash_guid (&d->key) ^ basehash;
return &d->c;
}
static bool serdata_builtin_eqkey(const struct ddsi_serdata *acmn, const struct ddsi_serdata *bcmn)
{
const struct ddsi_serdata_builtin *a = (const struct ddsi_serdata_builtin *)acmn;
const struct ddsi_serdata_builtin *b = (const struct ddsi_serdata_builtin *)bcmn;
return memcmp (&a->key, &b->key, sizeof (a->key)) == 0;
}
static void serdata_builtin_free(struct ddsi_serdata *dcmn)
{
struct ddsi_serdata_builtin *d = (struct ddsi_serdata_builtin *)dcmn;
if (d->c.kind == SDK_DATA)
nn_xqos_fini (&d->xqos);
os_free (d);
}
static struct ddsi_serdata_builtin *serdata_builtin_new(const struct ddsi_sertopic_builtin *tp, enum ddsi_serdata_kind kind)
{
struct ddsi_serdata_builtin *d = os_malloc(sizeof (*d));
ddsi_serdata_init (&d->c, &tp->c, kind);
return d;
}
static void from_entity_pp (struct ddsi_serdata_builtin *d, const struct participant *pp)
{
nn_xqos_copy(&d->xqos, &pp->plist->qos);
}
static void from_entity_proxypp (struct ddsi_serdata_builtin *d, const struct proxy_participant *proxypp)
{
nn_xqos_copy(&d->xqos, &proxypp->plist->qos);
}
static void set_topic_type_from_sertopic (struct ddsi_serdata_builtin *d, const struct ddsi_sertopic *tp)
{
if (!(d->xqos.present & QP_TOPIC_NAME))
{
d->xqos.topic_name = dds_string_dup (tp->name);
d->xqos.present |= QP_TOPIC_NAME;
}
if (!(d->xqos.present & QP_TYPE_NAME))
{
d->xqos.type_name = dds_string_dup (tp->typename);
d->xqos.present |= QP_TYPE_NAME;
}
}
static void from_entity_rd (struct ddsi_serdata_builtin *d, const struct reader *rd)
{
nn_xqos_copy(&d->xqos, rd->xqos);
set_topic_type_from_sertopic(d, rd->topic);
}
static void from_entity_prd (struct ddsi_serdata_builtin *d, const struct proxy_reader *prd)
{
nn_xqos_copy(&d->xqos, prd->c.xqos);
assert (d->xqos.present & QP_TOPIC_NAME);
assert (d->xqos.present & QP_TYPE_NAME);
}
static void from_entity_wr (struct ddsi_serdata_builtin *d, const struct writer *wr)
{
nn_xqos_copy(&d->xqos, wr->xqos);
set_topic_type_from_sertopic(d, wr->topic);
}
static void from_entity_pwr (struct ddsi_serdata_builtin *d, const struct proxy_writer *pwr)
{
nn_xqos_copy(&d->xqos, pwr->c.xqos);
assert (d->xqos.present & QP_TOPIC_NAME);
assert (d->xqos.present & QP_TYPE_NAME);
}
struct ddsi_serdata *ddsi_serdata_builtin_from_keyhash (const struct ddsi_sertopic *tpcmn, const nn_keyhash_t *keyhash)
{
/* FIXME: not quite elegant to manage the creation of a serdata for a built-in topic via this function, but I also find it quite unelegant to let from_sample read straight from the underlying internal entity, and to_sample convert to the external format ... I could claim the internal entity is the "serialised form", but that forces wrapping it in a fragchain in one way or another, which, though possible, is also a bit lacking in elegance. */
const struct ddsi_sertopic_builtin *tp = (const struct ddsi_sertopic_builtin *)tpcmn;
/* keyhash must in host format (which the GUIDs always are internally) */
const struct entity_common *entity = ephash_lookup_guid_untyped ((const nn_guid_t *) keyhash->value);
struct ddsi_serdata_builtin *d = serdata_builtin_new(tp, entity ? SDK_DATA : SDK_KEY);
memcpy (&d->key, keyhash->value, sizeof (d->key));
if (entity)
{
switch (entity->kind)
{
case EK_PARTICIPANT:
assert (tp->type == DSBT_PARTICIPANT);
from_entity_pp (d, (const struct participant *) entity);
break;
case EK_READER:
assert (tp->type == DSBT_READER);
from_entity_rd (d, (const struct reader *) entity);
break;
case EK_WRITER:
assert (tp->type == DSBT_WRITER);
from_entity_wr (d, (const struct writer *) entity);
break;
case EK_PROXY_PARTICIPANT:
assert (tp->type == DSBT_PARTICIPANT);
from_entity_proxypp (d, (const struct proxy_participant *) entity);
break;
case EK_PROXY_READER:
assert (tp->type == DSBT_READER);
from_entity_prd (d, (const struct proxy_reader *) entity);
break;
case EK_PROXY_WRITER:
assert (tp->type == DSBT_WRITER);
from_entity_pwr (d, (const struct proxy_writer *) entity);
break;
}
}
return fix_serdata_builtin(d, tp->c.serdata_basehash);
}
static struct ddsi_serdata *serdata_builtin_to_topicless (const struct ddsi_serdata *serdata_common)
{
/* All built-in ones are currently topicless */
return ddsi_serdata_ref (serdata_common);
}
static void convkey (dds_builtintopic_guid_t *key, const nn_guid_t *guid)
{
nn_guid_t tmp;
tmp = nn_hton_guid (*guid);
memcpy (key, &tmp, sizeof (*key));
}
static char *dds_string_dup_reuse (char *old, const char *src)
{
size_t size = strlen (src) + 1;
char *new = dds_realloc(old, size);
return memcpy (new, src, size);
}
static dds_qos_t *dds_qos_from_xqos_reuse (dds_qos_t *old, const nn_xqos_t *src)
{
if (old == NULL)
{
old = os_malloc (sizeof (*old));
nn_xqos_init_empty (old);
old->present |= QP_TOPIC_NAME | QP_TYPE_NAME;
nn_xqos_mergein_missing (old, src);
old->present &= ~(QP_TOPIC_NAME | QP_TYPE_NAME);
}
else
{
nn_xqos_fini (old);
nn_xqos_init_empty (old);
old->present |= QP_TOPIC_NAME | QP_TYPE_NAME;
nn_xqos_mergein_missing (old, src);
old->present &= ~(QP_TOPIC_NAME | QP_TYPE_NAME);
}
return old;
}
static bool to_sample_pp (const struct ddsi_serdata_builtin *d, struct dds_builtintopic_participant *sample)
{
convkey (&sample->key, &d->key);
if (d->c.kind == SDK_DATA)
{
sample->qos = dds_qos_from_xqos_reuse (sample->qos, &d->xqos);
}
return true;
}
static bool to_sample_endpoint (const struct ddsi_serdata_builtin *d, struct dds_builtintopic_endpoint *sample)
{
nn_guid_t ppguid;
convkey (&sample->key, &d->key);
ppguid = d->key;
ppguid.entityid.u = NN_ENTITYID_PARTICIPANT;
convkey (&sample->participant_key, &ppguid);
if (d->c.kind == SDK_DATA)
{
assert (d->xqos.present & QP_TOPIC_NAME);
assert (d->xqos.present & QP_TYPE_NAME);
sample->topic_name = dds_string_dup_reuse (sample->topic_name, d->xqos.topic_name);
sample->type_name = dds_string_dup_reuse (sample->type_name, d->xqos.type_name);
sample->qos = dds_qos_from_xqos_reuse (sample->qos, &d->xqos);
}
return true;
}
static bool serdata_builtin_topicless_to_sample (const struct ddsi_sertopic *topic, const struct ddsi_serdata *serdata_common, void *sample, void **bufptr, void *buflim)
{
const struct ddsi_serdata_builtin *d = (const struct ddsi_serdata_builtin *)serdata_common;
const struct ddsi_sertopic_builtin *tp = (const struct ddsi_sertopic_builtin *)topic;
if (bufptr) abort(); else { (void)buflim; } /* FIXME: haven't implemented that bit yet! */
/* FIXME: completing builtin topic support along these lines requires subscribers, publishers and topics to also become DDSI entities - which is probably a good thing anyway */
switch (tp->type)
{
case DSBT_PARTICIPANT:
return to_sample_pp (d, sample);
case DSBT_READER:
case DSBT_WRITER:
return to_sample_endpoint (d, sample);
}
assert (0);
return false;
}
static bool serdata_builtin_to_sample (const struct ddsi_serdata *serdata_common, void *sample, void **bufptr, void *buflim)
{
return serdata_builtin_topicless_to_sample (serdata_common->topic, serdata_common, sample, bufptr, buflim);
}
static uint32_t serdata_builtin_get_size (const struct ddsi_serdata *serdata_common)
{
(void)serdata_common;
return 0;
}
static void serdata_builtin_to_ser (const struct ddsi_serdata *serdata_common, size_t off, size_t sz, void *buf)
{
(void)serdata_common; (void)off; (void)sz; (void)buf;
}
static struct ddsi_serdata *serdata_builtin_to_ser_ref (const struct ddsi_serdata *serdata_common, size_t off, size_t sz, os_iovec_t *ref)
{
(void)serdata_common; (void)off; (void)sz; (void)ref;
return NULL;
}
static void serdata_builtin_to_ser_unref (struct ddsi_serdata *serdata_common, const os_iovec_t *ref)
{
(void)serdata_common; (void)ref;
}
const struct ddsi_serdata_ops ddsi_serdata_ops_builtin = {
.get_size = serdata_builtin_get_size,
.eqkey = serdata_builtin_eqkey,
.free = serdata_builtin_free,
.from_ser = 0,
.from_keyhash = ddsi_serdata_builtin_from_keyhash,
.from_sample = 0,
.to_ser = serdata_builtin_to_ser,
.to_sample = serdata_builtin_to_sample,
.to_ser_ref = serdata_builtin_to_ser_ref,
.to_ser_unref = serdata_builtin_to_ser_unref,
.to_topicless = serdata_builtin_to_topicless,
.topicless_to_sample = serdata_builtin_topicless_to_sample
};

147
src/core/ddsi/src/ddsi_sertopic_builtin.c
View file

@ -1,147 +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
*/
#include <stddef.h>
#include <ctype.h>
#include <assert.h>
#include <string.h>
#include "os/os.h"
#include "ddsi/q_md5.h"
#include "ddsi/q_bswap.h"
#include "ddsi/q_config.h"
#include "ddsi/q_freelist.h"
#include "ddsi/ddsi_sertopic.h"
#include "ddsi/ddsi_serdata_builtin.h"
#include "ddsc/dds.h"
/* FIXME: sertopic /= ddstopic so a lot of stuff needs to be moved here from dds_topic.c and the free function needs to be implemented properly */
struct ddsi_sertopic *new_sertopic_builtin (enum ddsi_sertopic_builtin_type type, const char *name, const char *typename)
{
struct ddsi_sertopic_builtin *tp = os_malloc (sizeof (*tp));
tp->c.iid = ddsi_iid_gen();
tp->c.name = dds_string_dup (name);
tp->c.typename = dds_string_dup (typename);
const size_t name_typename_size = strlen (tp->c.name) + 1 + strlen (tp->c.typename) + 1;
tp->c.name_typename = dds_alloc (name_typename_size);
snprintf (tp->c.name_typename, name_typename_size, "%s/%s", tp->c.name, tp->c.typename);
tp->c.ops = &ddsi_sertopic_ops_builtin;
tp->c.serdata_ops = &ddsi_serdata_ops_builtin;
tp->c.serdata_basehash = ddsi_sertopic_compute_serdata_basehash (tp->c.serdata_ops);
tp->c.status_cb = 0;
tp->c.status_cb_entity = NULL;
os_atomic_st32 (&tp->c.refc, 1);
tp->type = type;
return &tp->c;
}
static void sertopic_builtin_deinit (struct ddsi_sertopic *tp)
{
(void)tp;
}
static void free_pp (void *vsample)
{
dds_builtintopic_participant_t *sample = vsample;
dds_delete_qos (sample->qos);
sample->qos = NULL;
}
static void free_endpoint (void *vsample)
{
dds_builtintopic_endpoint_t *sample = vsample;
dds_free (sample->topic_name);
dds_free (sample->type_name);
dds_delete_qos (sample->qos);
sample->topic_name = sample->type_name = NULL;
sample->qos = NULL;
}
static size_t get_size (enum ddsi_sertopic_builtin_type type)
{
switch (type)
{
case DSBT_PARTICIPANT:
return sizeof (dds_builtintopic_participant_t);
case DSBT_READER:
case DSBT_WRITER:
return sizeof (dds_builtintopic_endpoint_t);
}
assert (0);
return 0;
}
static void sertopic_builtin_zero_samples (const struct ddsi_sertopic *sertopic_common, void *samples, size_t count)
{
const struct ddsi_sertopic_builtin *tp = (const struct ddsi_sertopic_builtin *)sertopic_common;
size_t size = get_size (tp->type);
memset (samples, 0, size * count);
}
static void sertopic_builtin_realloc_samples (void **ptrs, const struct ddsi_sertopic *sertopic_common, void *old, size_t oldcount, size_t count)
{
const struct ddsi_sertopic_builtin *tp = (const struct ddsi_sertopic_builtin *)sertopic_common;
const size_t size = get_size (tp->type);
char *new = dds_realloc (old, size * count);
if (new && count > oldcount)
memset (new + size * oldcount, 0, size * (count - oldcount));
for (size_t i = 0; i < count; i++)
{
void *ptr = (char *) new + i * size;
ptrs[i] = ptr;
}
}
static void sertopic_builtin_free_samples (const struct ddsi_sertopic *sertopic_common, void **ptrs, size_t count, dds_free_op_t op)
{
if (count > 0)
{
const struct ddsi_sertopic_builtin *tp = (const struct ddsi_sertopic_builtin *)sertopic_common;
const size_t size = get_size (tp->type);
#ifndef NDEBUG
for (size_t i = 0, off = 0; i < count; i++, off += size)
assert ((char *)ptrs[i] == (char *)ptrs[0] + off);
#endif
if (op & DDS_FREE_CONTENTS_BIT)
{
void (*f) (void *) = 0;
char *ptr = ptrs[0];
switch (tp->type)
{
case DSBT_PARTICIPANT:
f = free_pp;
break;
case DSBT_READER:
case DSBT_WRITER:
f = free_endpoint;
break;
}
assert (f != 0);
for (size_t i = 0; i < count; i++)
{
f (ptr);
ptr += size;
}
}
if (op & DDS_FREE_ALL_BIT)
{
dds_free (ptrs[0]);
}
}
}
const struct ddsi_sertopic_ops ddsi_sertopic_ops_builtin = {
.deinit = sertopic_builtin_deinit,
.zero_samples = sertopic_builtin_zero_samples,
.realloc_samples = sertopic_builtin_realloc_samples,
.free_samples = sertopic_builtin_free_samples
};

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

@ -149,39 +149,41 @@ int is_builtin_entityid (nn_entityid_t id, nn_vendorid_t vendorid)
}
}
static int is_builtin_endpoint (nn_entityid_t id, nn_vendorid_t vendorid)
int is_builtin_endpoint (nn_entityid_t id, nn_vendorid_t vendorid)
{
return is_builtin_entityid (id, vendorid) && id.u != NN_ENTITYID_PARTICIPANT;
}
static void entity_common_init (struct entity_common *e, const struct nn_guid *guid, const char *name, enum entity_kind kind, nn_vendorid_t vendorid, bool onlylocal, struct ddsi_tkmap_instance **tk)
bool is_local_orphan_endpoint (const struct entity_common *e)
{
return (e->guid.prefix.u[0] == 0 && e->guid.prefix.u[1] == 0 && e->guid.prefix.u[2] == 0 &&
is_builtin_endpoint (e->guid.entityid, ownvendorid));
}
static void entity_common_init (struct entity_common *e, const struct nn_guid *guid, const char *name, enum entity_kind kind, nn_wctime_t tcreate, nn_vendorid_t vendorid, bool onlylocal)
{
e->guid = *guid;
e->kind = kind;
e->tupdate = tcreate;
e->name = os_strdup (name ? name : "");
e->onlylocal = onlylocal;
os_mutexInit (&e->lock);
if (onlylocal || is_builtin_endpoint (guid->entityid, vendorid))
if (ddsi_plugin.builtintopic_is_visible (guid->entityid, onlylocal, vendorid))
{
e->iid = ddsi_iid_gen ();
*tk = NULL;
e->tk = ddsi_plugin.builtintopic_get_tkmap_entry (guid);
e->iid = e->tk->m_iid;
}
else
{
struct ddsi_serdata *sd;
struct nn_keyhash kh;
memcpy (&kh, guid, sizeof (kh));
/* any random builtin topic will do (provided it has a GUID for a key), because what matters is the "class" of the topic, not the actual topic; also, this is called early in the initialisation of the entity with this GUID, which simply causes serdata_from_keyhash to create a key-only serdata because the key lookup fails. */
sd = ddsi_serdata_from_keyhash (gv.builtin_participant_topic, &kh);
/* FIXME: this makes the iid for a reincarnation of a proxy entity dependent on whether an application reader kept the corresponding built-in topic instance around, it may be attractive to reconsider and guarantee a new iid in these cases, at least for the publication handle */
*tk = ddsi_tkmap_find(sd, false, true);
ddsi_serdata_unref (sd);
e->iid = (*tk)->m_iid;
e->tk = NULL;
e->iid = ddsi_iid_gen ();
}
}
static void entity_common_fini (struct entity_common *e)
{
if (e->tk)
ddsi_tkmap_instance_unref (e->tk);
os_free (e->name);
os_mutexDestroy (&e->lock);
}
@ -238,38 +240,27 @@ void local_reader_ary_setinvalid (struct local_reader_ary *x)
os_mutexUnlock (&x->rdary_lock);
}
static void write_builtin_topic_any (const struct entity_common *e, nn_wctime_t timestamp, bool alive, nn_vendorid_t vendorid, struct ddsi_tkmap_instance *tk)
nn_vendorid_t get_entity_vendorid (const struct entity_common *e)
{
if (!(e->onlylocal || is_builtin_endpoint(e->guid.entityid, vendorid)))
switch (e->kind)
{
/* initialize to avoid gcc warning ultimately caused by C's horrible type system */
enum ddsi_sertopic_builtin_type type = DSBT_PARTICIPANT;
switch (e->kind)
{
case EK_PARTICIPANT:
case EK_PROXY_PARTICIPANT:
type = DSBT_PARTICIPANT;
break;
case EK_READER:
case EK_PROXY_READER:
type = DSBT_READER;
break;
case EK_WRITER:
case EK_PROXY_WRITER:
type = DSBT_WRITER;
break;
}
assert(type != DSBT_PARTICIPANT || (e->kind == EK_PARTICIPANT || e->kind == EK_PROXY_PARTICIPANT));
ddsi_plugin.builtin_write (type, &e->guid, timestamp, alive);
case EK_PARTICIPANT:
case EK_READER:
case EK_WRITER:
return (nn_vendorid_t) MY_VENDOR_ID;
break;
case EK_PROXY_PARTICIPANT:
return ((const struct proxy_participant *) e)->vendor;
break;
case EK_PROXY_READER:
return ((const struct proxy_reader *) e)->c.vendor;
break;
case EK_PROXY_WRITER:
return ((const struct proxy_writer *) e)->c.vendor;
break;
}
/* tkmap instance only needs to be kept around until the first write of a built-in topic (if none ever happens, it needn't be kept at all): afterward, the WHC of the local built-in topic writer will keep the entry alive. FIXME: the SPDP/SEPD ones currently use default sertopics instead of builtin sertopics, and so use different mappings and different instnace ids. No-one ever sees those ids, so it doesn't matter, but it would nicer if it could actually be the same one. FIXME: it would also be nicer if the local built-in topics and the SPDP/SEDP writers were the same, but I want the locally created endpoints visible in the built-in topics as well, and those don't exist in the discovery writers ... */
if (tk)
ddsi_tkmap_instance_unref (tk);
}
static void write_builtin_topic_local (const struct entity_common *e, nn_wctime_t timestamp, bool alive, struct ddsi_tkmap_instance *tk)
{
write_builtin_topic_any(e, timestamp, alive, ownvendorid, tk);
assert (0);
return (nn_vendorid_t) NN_VENDORID_UNKNOWN;
}
/* DELETED PARTICIPANTS --------------------------------------------- */
@ -408,7 +399,6 @@ int new_participant_guid (const nn_guid_t *ppguid, unsigned flags, const nn_plis
{
struct participant *pp;
nn_guid_t subguid, group_guid;
struct ddsi_tkmap_instance *tk;
/* no reserved bits may be set */
assert ((flags & ~(RTPS_PF_NO_BUILTIN_READERS | RTPS_PF_NO_BUILTIN_WRITERS | RTPS_PF_PRIVILEGED_PP | RTPS_PF_IS_DDSI2_PP | RTPS_PF_ONLY_LOCAL)) == 0);
@ -451,7 +441,7 @@ int new_participant_guid (const nn_guid_t *ppguid, unsigned flags, const nn_plis
pp = os_malloc (sizeof (*pp));
entity_common_init (&pp->e, ppguid, "", EK_PARTICIPANT, ownvendorid, ((flags & RTPS_PF_ONLY_LOCAL) != 0), &tk);
entity_common_init (&pp->e, ppguid, "", EK_PARTICIPANT, now (), ownvendorid, ((flags & RTPS_PF_ONLY_LOCAL) != 0));
pp->user_refc = 1;
pp->builtin_refc = 0;
pp->builtins_deleted = 0;
@ -630,7 +620,7 @@ int new_participant_guid (const nn_guid_t *ppguid, unsigned flags, const nn_plis
trigger_recv_threads ();
}
write_builtin_topic_local(&pp->e, now(), true, tk);
ddsi_plugin.builtintopic_write (&pp->e, now(), true);
/* SPDP periodic broadcast uses the retransmit path, so the initial
publication must be done differently. Must be later than making
@ -866,7 +856,7 @@ int delete_participant (const struct nn_guid *ppguid)
struct participant *pp;
if ((pp = ephash_lookup_participant_guid (ppguid)) == NULL)
return ERR_UNKNOWN_ENTITY;
write_builtin_topic_local(&pp->e, now(), false, NULL);
ddsi_plugin.builtintopic_write (&pp->e, now(), false);
remember_deleted_participant_guid (&pp->e.guid);
ephash_remove_participant_guid (pp);
gcreq_participant (pp);
@ -2077,7 +2067,7 @@ static void connect_writer_with_reader (struct writer *wr, struct reader *rd, nn
{
int32_t reason;
(void)tnow;
if (is_builtin_entityid (wr->e.guid.entityid, ownvendorid) || is_builtin_entityid (rd->e.guid.entityid, ownvendorid))
if (!is_local_orphan_endpoint (&wr->e) && (is_builtin_entityid (wr->e.guid.entityid, ownvendorid) || is_builtin_entityid (rd->e.guid.entityid, ownvendorid)))
return;
if ((reason = qos_match_p (rd->xqos, wr->xqos)) >= 0)
{
@ -2295,7 +2285,7 @@ static void generic_do_local_match (struct entity_common *e, nn_mtime_t tnow)
struct ephash_enum est;
struct entity_common *em;
enum entity_kind mkind;
if (is_builtin_entityid (e->guid.entityid, ownvendorid))
if (is_builtin_entityid (e->guid.entityid, ownvendorid) && !is_local_orphan_endpoint (e))
/* never a need for local matches on discovery endpoints */
return;
mkind = generic_do_local_match_mkind(e->kind);
@ -2374,34 +2364,27 @@ static void new_reader_writer_common (const struct nn_guid *guid, const struct d
topic ? topic->typename : "(null)");
}
static void endpoint_common_init
(
struct entity_common *e,
struct endpoint_common *c,
enum entity_kind kind,
const struct nn_guid *guid,
const struct nn_guid *group_guid,
struct participant *pp,
struct ddsi_tkmap_instance **tk
)
static void endpoint_common_init (struct entity_common *e, struct endpoint_common *c, enum entity_kind kind, const struct nn_guid *guid, const struct nn_guid *group_guid, struct participant *pp)
{
entity_common_init (e, guid, NULL, kind, ownvendorid, pp->e.onlylocal, tk);
entity_common_init (e, guid, NULL, kind, now (), ownvendorid, pp->e.onlylocal);
c->pp = ref_participant (pp, &e->guid);
if (group_guid)
{
c->group_guid = *group_guid;
}
else
{
memset (&c->group_guid, 0, sizeof (c->group_guid));
}
}
static void endpoint_common_fini (struct entity_common *e, struct endpoint_common *c)
{
if (!is_builtin_entityid(e->guid.entityid, ownvendorid))
pp_release_entityid(c->pp, e->guid.entityid);
unref_participant (c->pp, &e->guid);
if (c->pp)
unref_participant (c->pp, &e->guid);
else
{
/* only for the (almost pseudo) writers used for generating the built-in topics */
assert (is_local_orphan_endpoint (e));
}
entity_common_fini (e);
}
@ -2608,25 +2591,8 @@ unsigned remove_acked_messages (struct writer *wr, struct whc_state *whcst, stru
return n;
}
static struct writer * new_writer_guid (const struct nn_guid *guid, const struct nn_guid *group_guid, struct participant *pp, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc *whc, status_cb_t status_cb, void * status_entity)
static void new_writer_guid_common_init (struct writer *wr, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc *whc, status_cb_t status_cb, void * status_entity)
{
struct writer *wr;
nn_mtime_t tnow = now_mt ();
struct ddsi_tkmap_instance *tk;
assert (is_writer_entityid (guid->entityid));
assert (ephash_lookup_writer_guid (guid) == NULL);
assert (memcmp (&guid->prefix, &pp->e.guid.prefix, sizeof (guid->prefix)) == 0);
new_reader_writer_common (guid, topic, xqos);
wr = os_malloc (sizeof (*wr));
/* want a pointer to the participant so that a parallel call to
delete_participant won't interfere with our ability to address
the participant */
endpoint_common_init (&wr->e, &wr->c, EK_WRITER, guid, group_guid, pp, &tk);
os_condInit (&wr->throttle_cond, &wr->e.lock);
wr->seq = 0;
wr->cs_seq = 0;
@ -2658,12 +2624,10 @@ static struct writer * new_writer_guid (const struct nn_guid *guid, const struct
assert (wr->xqos->aliased == 0);
set_topic_type_name (wr->xqos, topic);
if (dds_get_log_mask() & DDS_LC_DISCOVERY)
{
DDS_LOG(DDS_LC_DISCOVERY, "WRITER %x:%x:%x:%x QOS={", PGUID (wr->e.guid));
nn_log_xqos (DDS_LC_DISCOVERY, wr->xqos);
DDS_LOG(DDS_LC_DISCOVERY, "}\n");
}
DDS_LOG(DDS_LC_DISCOVERY, "WRITER %x:%x:%x:%x QOS={", PGUID (wr->e.guid));
nn_log_xqos (DDS_LC_DISCOVERY, wr->xqos);
DDS_LOG(DDS_LC_DISCOVERY, "}\n");
assert (wr->xqos->present & QP_RELIABILITY);
wr->reliable = (wr->xqos->reliability.kind != NN_BEST_EFFORT_RELIABILITY_QOS);
assert (wr->xqos->present & QP_DURABILITY);
@ -2818,23 +2782,43 @@ static struct writer * new_writer_guid (const struct nn_guid *guid, const struct
ut_avlInit (&wr_local_readers_treedef, &wr->local_readers);
local_reader_ary_init (&wr->rdary);
}
static struct writer *new_writer_guid (const struct nn_guid *guid, const struct nn_guid *group_guid, struct participant *pp, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc *whc, status_cb_t status_cb, void *status_entity)
{
struct writer *wr;
nn_mtime_t tnow = now_mt ();
assert (is_writer_entityid (guid->entityid));
assert (ephash_lookup_writer_guid (guid) == NULL);
assert (memcmp (&guid->prefix, &pp->e.guid.prefix, sizeof (guid->prefix)) == 0);
new_reader_writer_common (guid, topic, xqos);
wr = os_malloc (sizeof (*wr));
/* want a pointer to the participant so that a parallel call to
delete_participant won't interfere with our ability to address
the participant */
endpoint_common_init (&wr->e, &wr->c, EK_WRITER, guid, group_guid, pp);
new_writer_guid_common_init(wr, topic, xqos, whc, status_cb, status_entity);
/* guid_hash needed for protocol handling, so add it before we send
out our first message. Also: needed for matching, and swapping
the order if hash insert & matching creates a window during which
neither of two endpoints being created in parallel can discover
the other. */
out our first message. Also: needed for matching, and swapping
the order if hash insert & matching creates a window during which
neither of two endpoints being created in parallel can discover
the other. */
ephash_insert_writer_guid (wr);
/* once it exists, match it with proxy writers and broadcast
existence (I don't think it matters much what the order of these
two is, but it seems likely that match-then-broadcast has a
slightly lower likelihood that a response from a proxy reader
gets dropped) -- but note that without adding a lock it might be
deleted while we do so */
existence (I don't think it matters much what the order of these
two is, but it seems likely that match-then-broadcast has a
slightly lower likelihood that a response from a proxy reader
gets dropped) -- but note that without adding a lock it might be
deleted while we do so */
match_writer_with_proxy_readers (wr, tnow);
match_writer_with_local_readers (wr, tnow);
write_builtin_topic_local(&wr->e, now(), true, tk);
ddsi_plugin.builtintopic_write (&wr->e, now(), true);
sedp_write_writer (wr);
if (wr->lease_duration != T_NEVER)
@ -2846,7 +2830,7 @@ static struct writer * new_writer_guid (const struct nn_guid *guid, const struct
return wr;
}
struct writer * new_writer (struct nn_guid *wrguid, const struct nn_guid *group_guid, const struct nn_guid *ppguid, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc * whc, status_cb_t status_cb, void * status_cb_arg)
struct writer *new_writer (struct nn_guid *wrguid, const struct nn_guid *group_guid, const struct nn_guid *ppguid, const struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc * whc, status_cb_t status_cb, void *status_cb_arg)
{
struct participant *pp;
struct writer * wr;
@ -2866,6 +2850,29 @@ struct writer * new_writer (struct nn_guid *wrguid, const struct nn_guid *group_
return wr;
}
struct local_orphan_writer *new_local_orphan_writer (nn_entityid_t entityid, struct ddsi_sertopic *topic, const struct nn_xqos *xqos, struct whc *whc)
{
nn_guid_t guid;
struct local_orphan_writer *lowr;
struct writer *wr;
nn_mtime_t tnow = now_mt ();
DDS_LOG(DDS_LC_DISCOVERY, "new_local_orphan_writer(%s/%s)\n", topic->name, topic->typename);
lowr = os_malloc (sizeof (*lowr));
wr = &lowr->wr;
memset (&guid.prefix, 0, sizeof (guid.prefix));
guid.entityid = entityid;
entity_common_init (&wr->e, &guid, NULL, EK_WRITER, now (), ownvendorid, true);
wr->c.pp = NULL;
memset (&wr->c.group_guid, 0, sizeof (wr->c.group_guid));
new_writer_guid_common_init (wr, topic, xqos, whc, 0, NULL);
ephash_insert_writer_guid (wr);
match_writer_with_local_readers (wr, tnow);
ddsi_plugin.builtintopic_write (&wr->e, now(), true);
return lowr;
}
static void gc_delete_writer (struct gcreq *gcreq)
{
struct writer *wr = gcreq->arg;
@ -2960,7 +2967,7 @@ int delete_writer_nolinger_locked (struct writer *wr)
{
DDS_LOG(DDS_LC_DISCOVERY, "delete_writer_nolinger(guid %x:%x:%x:%x) ...\n", PGUID (wr->e.guid));
ASSERT_MUTEX_HELD (&wr->e.lock);
write_builtin_topic_local(&wr->e, now(), false, NULL);
ddsi_plugin.builtintopic_write (&wr->e, now(), false);
local_reader_ary_setinvalid (&wr->rdary);
ephash_remove_writer_guid (wr);
writer_set_state (wr, WRST_DELETING);
@ -2990,6 +2997,13 @@ int delete_writer_nolinger (const struct nn_guid *guid)
return 0;
}
void delete_local_orphan_writer (struct local_orphan_writer *lowr)
{
os_mutexLock (&lowr->wr.e.lock);
delete_writer_nolinger_locked (&lowr->wr);
os_mutexUnlock (&lowr->wr.e.lock);
}
int delete_writer (const struct nn_guid *guid)
{
struct writer *wr;
@ -3177,7 +3191,6 @@ static struct reader * new_reader_guid
struct reader * rd;
nn_mtime_t tnow = now_mt ();
struct ddsi_tkmap_instance *tk;
assert (!is_writer_entityid (guid->entityid));
assert (ephash_lookup_reader_guid (guid) == NULL);
@ -3186,7 +3199,7 @@ static struct reader * new_reader_guid
new_reader_writer_common (guid, topic, xqos);
rd = os_malloc (sizeof (*rd));
endpoint_common_init (&rd->e, &rd->c, EK_READER, guid, group_guid, pp, &tk);
endpoint_common_init (&rd->e, &rd->c, EK_READER, guid, group_guid, pp);
/* Copy QoS, merging in defaults */
rd->xqos = os_malloc (sizeof (*rd->xqos));
@ -3290,7 +3303,7 @@ static struct reader * new_reader_guid
ephash_insert_reader_guid (rd);
match_reader_with_proxy_writers (rd, tnow);
match_reader_with_local_writers (rd, tnow);
write_builtin_topic_local(&rd->e, now(), true, tk);
ddsi_plugin.builtintopic_write (&rd->e, now(), true);
sedp_write_reader (rd);
return rd;
}
@ -3383,7 +3396,7 @@ int delete_reader (const struct nn_guid *guid)
(ddsi_plugin.rhc_plugin.rhc_fini_fn) (rd->rhc);
}
DDS_LOG(DDS_LC_DISCOVERY, "delete_reader_guid(guid %x:%x:%x:%x) ...\n", PGUID (*guid));
write_builtin_topic_local(&rd->e, now(), false, NULL);
ddsi_plugin.builtintopic_write (&rd->e, now(), false);
ephash_remove_reader_guid (rd);
gcreq_reader (rd);
return 0;
@ -3459,7 +3472,6 @@ void new_proxy_participant
runs on a single thread, it can't go wrong. FIXME, maybe? The
same holds for the other functions for creating entities. */
struct proxy_participant *proxypp;
struct ddsi_tkmap_instance *tk;
assert (ppguid->entityid.u == NN_ENTITYID_PARTICIPANT);
assert (ephash_lookup_proxy_participant_guid (ppguid) == NULL);
@ -3469,7 +3481,7 @@ void new_proxy_participant
proxypp = os_malloc (sizeof (*proxypp));
entity_common_init (&proxypp->e, ppguid, "", EK_PROXY_PARTICIPANT, vendor, false, &tk);
entity_common_init (&proxypp->e, ppguid, "", EK_PROXY_PARTICIPANT, timestamp, vendor, false);
proxypp->refc = 1;
proxypp->lease_expired = 0;
proxypp->vendor = vendor;
@ -3625,7 +3637,7 @@ void new_proxy_participant
if (proxypp->owns_lease)
lease_register (os_atomic_ldvoidp (&proxypp->lease));
write_builtin_topic_any(&proxypp->e, timestamp, true, proxypp->vendor, tk);
ddsi_plugin.builtintopic_write (&proxypp->e, timestamp, true);
os_mutexUnlock (&proxypp->e.lock);
}
@ -3643,7 +3655,7 @@ int update_proxy_participant_plist_locked (struct proxy_participant *proxypp, co
switch (source)
{
case UPD_PROXYPP_SPDP:
write_builtin_topic_any(&proxypp->e, timestamp, true, proxypp->vendor, NULL);
ddsi_plugin.builtintopic_write (&proxypp->e, timestamp, true);
proxypp->proxypp_have_spdp = 1;
break;
case UPD_PROXYPP_CM:
@ -3892,7 +3904,7 @@ int delete_proxy_participant_by_guid (const struct nn_guid * guid, nn_wctime_t t
return ERR_UNKNOWN_ENTITY;
}
DDS_LOG(DDS_LC_DISCOVERY, "- deleting\n");
write_builtin_topic_any(&ppt->e, timestamp, false, ppt->vendor, NULL);
ddsi_plugin.builtintopic_write (&ppt->e, timestamp, false);
remember_deleted_participant_guid (&ppt->e.guid);
ephash_remove_proxy_participant_guid (ppt);
os_mutexUnlock (&gv.lock);
@ -4021,12 +4033,7 @@ void delete_proxy_group (const nn_guid_t *guid, nn_wctime_t timestamp, int isimp
/* PROXY-ENDPOINT --------------------------------------------------- */
static void proxy_endpoint_common_init
(
struct entity_common *e, struct proxy_endpoint_common *c,
enum entity_kind kind, const struct nn_guid *guid, struct proxy_participant *proxypp,
struct addrset *as, const nn_plist_t *plist, struct ddsi_tkmap_instance **tk
)
static void proxy_endpoint_common_init (struct entity_common *e, struct proxy_endpoint_common *c, enum entity_kind kind, const struct nn_guid *guid, nn_wctime_t tcreate, struct proxy_participant *proxypp, struct addrset *as, const nn_plist_t *plist)
{
const char *name;
@ -4036,7 +4043,7 @@ static void proxy_endpoint_common_init
assert ((plist->qos.present & (QP_TOPIC_NAME | QP_TYPE_NAME)) == (QP_TOPIC_NAME | QP_TYPE_NAME));
name = (plist->present & PP_ENTITY_NAME) ? plist->entity_name : "";
entity_common_init (e, guid, name, kind, proxypp->vendor, false, tk);
entity_common_init (e, guid, name, kind, tcreate, proxypp->vendor, false);
c->xqos = nn_xqos_dup (&plist->qos);
c->as = ref_addrset (as);
c->topic = NULL; /* set from first matching reader/writer */
@ -4071,8 +4078,7 @@ int new_proxy_writer (const struct nn_guid *ppguid, const struct nn_guid *guid,
struct proxy_writer *pwr;
int isreliable;
nn_mtime_t tnow = now_mt ();
struct ddsi_tkmap_instance *tk;
(void)timestamp;
assert (is_writer_entityid (guid->entityid));
assert (ephash_lookup_proxy_writer_guid (guid) == NULL);
@ -4083,7 +4089,7 @@ int new_proxy_writer (const struct nn_guid *ppguid, const struct nn_guid *guid,
}
pwr = os_malloc (sizeof (*pwr));
proxy_endpoint_common_init (&pwr->e, &pwr->c, EK_PROXY_WRITER, guid, proxypp, as, plist, &tk);
proxy_endpoint_common_init (&pwr->e, &pwr->c, EK_PROXY_WRITER, guid, timestamp, proxypp, as, plist);
ut_avlInit (&pwr_readers_treedef, &pwr->readers);
pwr->n_reliable_readers = 0;
@ -4149,7 +4155,7 @@ int new_proxy_writer (const struct nn_guid *ppguid, const struct nn_guid *guid,
local_reader_ary_init (&pwr->rdary);
ephash_insert_proxy_writer_guid (pwr);
match_proxy_writer_with_readers (pwr, tnow);
write_builtin_topic_any(&pwr->e, timestamp, true, pwr->c.vendor, tk);
ddsi_plugin.builtintopic_write (&pwr->e, timestamp, true);
os_mutexLock (&pwr->e.lock);
pwr->local_matching_inprogress = 0;
@ -4265,7 +4271,6 @@ static void gc_delete_proxy_writer (struct gcreq *gcreq)
int delete_proxy_writer (const struct nn_guid *guid, nn_wctime_t timestamp, int isimplicit)
{
struct proxy_writer *pwr;
(void)timestamp;
(void)isimplicit;
DDS_LOG(DDS_LC_DISCOVERY, "delete_proxy_writer (%x:%x:%x:%x) ", PGUID (*guid));
os_mutexLock (&gv.lock);
@ -4281,7 +4286,7 @@ int delete_proxy_writer (const struct nn_guid *guid, nn_wctime_t timestamp, int
from removing themselves from the proxy writer's rdary[]. */
local_reader_ary_setinvalid (&pwr->rdary);
DDS_LOG(DDS_LC_DISCOVERY, "- deleting\n");
write_builtin_topic_any(&pwr->e, timestamp, false, pwr->c.vendor, NULL);
ddsi_plugin.builtintopic_write (&pwr->e, timestamp, false);
ephash_remove_proxy_writer_guid (pwr);
os_mutexUnlock (&gv.lock);
gcreq_proxy_writer (pwr);
@ -4299,8 +4304,6 @@ int new_proxy_reader (const struct nn_guid *ppguid, const struct nn_guid *guid,
struct proxy_participant *proxypp;
struct proxy_reader *prd;
nn_mtime_t tnow = now_mt ();
struct ddsi_tkmap_instance *tk;
(void)timestamp;
assert (!is_writer_entityid (guid->entityid));
assert (ephash_lookup_proxy_reader_guid (guid) == NULL);
@ -4312,7 +4315,7 @@ int new_proxy_reader (const struct nn_guid *ppguid, const struct nn_guid *guid,
}
prd = os_malloc (sizeof (*prd));
proxy_endpoint_common_init (&prd->e, &prd->c, EK_PROXY_READER, guid, proxypp, as, plist, &tk);
proxy_endpoint_common_init (&prd->e, &prd->c, EK_PROXY_READER, guid, timestamp, proxypp, as, plist);
prd->deleting = 0;
#ifdef DDSI_INCLUDE_SSM
@ -4326,7 +4329,7 @@ int new_proxy_reader (const struct nn_guid *ppguid, const struct nn_guid *guid,
ut_avlInit (&prd_writers_treedef, &prd->writers);
ephash_insert_proxy_reader_guid (prd);
match_proxy_reader_with_writers (prd, tnow);
write_builtin_topic_any(&prd->e, timestamp, true, prd->c.vendor, tk);
ddsi_plugin.builtintopic_write (&prd->e, timestamp, true);
return 0;
}
@ -4399,7 +4402,6 @@ static void gc_delete_proxy_reader (struct gcreq *gcreq)
int delete_proxy_reader (const struct nn_guid *guid, nn_wctime_t timestamp, int isimplicit)
{
struct proxy_reader *prd;
(void)timestamp;
(void)isimplicit;
DDS_LOG(DDS_LC_DISCOVERY, "delete_proxy_reader (%x:%x:%x:%x) ", PGUID (*guid));
os_mutexLock (&gv.lock);
@ -4409,7 +4411,7 @@ int delete_proxy_reader (const struct nn_guid *guid, nn_wctime_t timestamp, int
DDS_LOG(DDS_LC_DISCOVERY, "- unknown\n");
return ERR_UNKNOWN_ENTITY;
}
write_builtin_topic_any(&prd->e, timestamp, false, prd->c.vendor, NULL);
ddsi_plugin.builtintopic_write (&prd->e, timestamp, false);
ephash_remove_proxy_reader_guid (prd);
os_mutexUnlock (&gv.lock);
DDS_LOG(DDS_LC_DISCOVERY, "- deleting\n");

13
src/core/ddsi/src/q_gc.c
View file

@ -180,6 +180,14 @@ struct gcreq_queue *gcreq_queue_new (void)
return q;
}
void gcreq_queue_drain (struct gcreq_queue *q)
{
os_mutexLock (&q->lock);
while (q->count != 0)
os_condWait (&q->cond, &q->lock);
os_mutexUnlock (&q->lock);
}
void gcreq_queue_free (struct gcreq_queue *q)
{
struct gcreq *gcreq;
@ -191,7 +199,8 @@ void gcreq_queue_free (struct gcreq_queue *q)
os_mutexLock (&q->lock);
q->terminate = 1;
/* Wait until there is only request in existence, the one we just
allocated. Then we know the gc system is quiet. */
allocated (this is also why we can't use "drain" here). Then
we know the gc system is quiet. */
while (q->count != 1)
os_condWait (&q->cond, &q->lock);
os_mutexUnlock (&q->lock);
@ -227,7 +236,7 @@ void gcreq_free (struct gcreq *gcreq)
struct gcreq_queue *gcreq_queue = gcreq->queue;
os_mutexLock (&gcreq_queue->lock);
--gcreq_queue->count;
if (gcreq_queue->terminate && gcreq_queue->count <= 1)
if (gcreq_queue->count <= 1)
os_condBroadcast (&gcreq_queue->cond);
os_mutexUnlock (&gcreq_queue->lock);
os_free (gcreq);

20
src/core/ddsi/src/q_init.c
View file

@ -54,7 +54,6 @@
#include "ddsi/ddsi_raweth.h"
#include "ddsi/ddsi_mcgroup.h"
#include "ddsi/ddsi_serdata_default.h"
#include "ddsi/ddsi_serdata_builtin.h"
#include "ddsi/ddsi_tkmap.h"
#include "dds__whc.h"
@ -772,18 +771,12 @@ static void make_special_topics (void)
{
gv.plist_topic = make_special_topic (PLATFORM_IS_LITTLE_ENDIAN ? PL_CDR_LE : PL_CDR_BE, &ddsi_serdata_ops_plist);
gv.rawcdr_topic = make_special_topic (PLATFORM_IS_LITTLE_ENDIAN ? CDR_LE : CDR_BE, &ddsi_serdata_ops_rawcdr);
gv.builtin_participant_topic = new_sertopic_builtin (DSBT_PARTICIPANT, "DCPSParticipant", "org::eclipse::cyclonedds::builtin::DCPSParticipant");
gv.builtin_reader_topic = new_sertopic_builtin (DSBT_READER, "DCPSSubscription", "org::eclipse::cyclonedds::builtin::DCPSSubscription");
gv.builtin_writer_topic = new_sertopic_builtin (DSBT_WRITER, "DCPSPublication", "org::eclipse::cyclonedds::builtin::DCPSPublication");
}
static void free_special_topics (void)
{
ddsi_sertopic_unref (gv.plist_topic);
ddsi_sertopic_unref (gv.rawcdr_topic);
ddsi_sertopic_unref (gv.builtin_participant_topic);
ddsi_sertopic_unref (gv.builtin_reader_topic);
ddsi_sertopic_unref (gv.builtin_writer_topic);
}
static int setup_and_start_recv_threads (void)
@ -1408,7 +1401,7 @@ static void builtins_dqueue_ready_cb (void *varg)
os_mutexUnlock (&arg->lock);
}
void rtps_term (void)
void rtps_stop (void)
{
struct thread_state1 *self = lookup_thread_state ();
#ifdef DDSI_INCLUDE_NETWORK_CHANNELS
@ -1527,17 +1520,26 @@ void rtps_term (void)
}
/* Wait until all participants are really gone => by then we can be
certain that no new GC requests will be added */
certain that no new GC requests will be added, short of what we
do here */
os_mutexLock (&gv.participant_set_lock);
while (gv.nparticipants > 0)
os_condWait (&gv.participant_set_cond, &gv.participant_set_lock);
os_mutexUnlock (&gv.participant_set_lock);
/* Wait until no more GC requests are outstanding -- not really
necessary, but it allows us to claim the stack is quiescent
at this point */
gcreq_queue_drain (gv.gcreq_queue);
/* Clean up privileged_pp -- it must be NULL now (all participants
are gone), but the lock still needs to be destroyed */
assert (gv.privileged_pp == NULL);
os_mutexDestroy (&gv.privileged_pp_lock);
}
void rtps_fini (void)
{
/* Shut down the GC system -- no new requests will be added */
gcreq_queue_free (gv.gcreq_queue);

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

@ -893,7 +893,7 @@ static int insert_sample_in_whc (struct writer *wr, seqno_t seq, struct nn_plist
res = 1;
#ifndef NDEBUG
if (wr->e.guid.entityid.u == NN_ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER)
if (wr->e.guid.entityid.u == NN_ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER && !is_local_orphan_endpoint (&wr->e))
{
struct whc_state whcst;
whc_get_state(wr->whc, &whcst);