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

Add DOMAIN_ID, DOMAIN_TAG to config, discovery

Browse source 
This commits adds support for the DOMAIN_ID and DOMAIN_TAG parameters in
participant discovery, allowing multiple domains to share a port
number (a feature introduced in DDSI 2.3).  The tag can be configured
via Discovery/Tag.

This commit also introduces a setting Discovery/ExternalDomainId that
makes it possible to override the domain id on the network, both in what
is advertised in the DOMAIN_ID discovery parameter and in the
calculation of port numbers.  This way a single process can create two
independent domains that talk via the network, which is on occassion
useful in writing tests.

Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson 2019-11-19 21:25:17 +01:00 • committed by eboasson
parent 15179910d1
commit 08d9c296f1
13 changed files with 130 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -37,8 +37,10 @@ struct ddsi_portmapping {
uint32_t d3;
};
bool ddsi_valid_portmapping (const struct ddsi_portmapping *map, uint32_t domain_id, int32_t participant_index, char *msg, size_t msgsize);
uint32_t ddsi_get_port (const struct ddsi_portmapping *map, enum ddsi_port which, uint32_t domain_id, int32_t participant_index);
struct config;
bool ddsi_valid_portmapping (const struct config *config, int32_t participant_index, char *msg, size_t msgsize);
uint32_t ddsi_get_port (const struct config *config, enum ddsi_port which, int32_t participant_index);
#if defined (__cplusplus)
}

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

@ -196,6 +196,8 @@ struct config
int dontRoute;
int enableMulticastLoopback;
uint32_t domainId;
struct config_maybe_uint32 extDomainId; // domain id advertised in discovery
char *domainTag;
int participantIndex;
int maxAutoParticipantIndex;
char *spdpMulticastAddressString;

4
src/core/ddsi/include/dds/ddsi/q_plist.h
View file

@ -59,6 +59,8 @@ extern "C" {
/* Security extensions. */
#define PP_IDENTITY_TOKEN ((uint64_t)1 << 41)
#define PP_PERMISSIONS_TOKEN ((uint64_t)1 << 42)
#define PP_DOMAIN_ID ((uint64_t)1 << 43)
#define PP_DOMAIN_TAG ((uint64_t)1 << 44)
/* Set for unrecognized parameters that are in the reserved space or
in our own vendor-specific space that have the
PID_UNRECOGNIZED_INCOMPATIBLE_FLAG set (see DDSI 2.1 9.6.2.2.1) */
@ -162,6 +164,8 @@ typedef struct nn_plist {
#ifdef DDSI_INCLUDE_SSM
nn_reader_favours_ssm_t reader_favours_ssm;
#endif
uint32_t domain_id;
char *domain_tag;
} nn_plist_t;

2
src/core/ddsi/include/dds/ddsi/q_protocol.h
View file

@ -330,6 +330,8 @@ DDSRT_WARNING_MSVC_ON(4200)
#define PID_PAD 0x0u
#define PID_SENTINEL 0x1u
#define PID_DOMAIN_ID 0xfu
#define PID_DOMAIN_TAG (0x14u | PID_UNRECOGNIZED_INCOMPATIBLE_FLAG)
#define PID_USER_DATA 0x2cu
#define PID_TOPIC_NAME 0x5u
#define PID_TYPE_NAME 0x7u

8
src/core/ddsi/src/ddsi_portmapping.c
View file

@ -89,7 +89,7 @@ static const char *portname (enum ddsi_port which)
return n;
}
bool ddsi_valid_portmapping (const struct ddsi_portmapping *map, uint32_t domain_id, int32_t participant_index, char *msg, size_t msgsize)
bool ddsi_valid_portmapping (const struct config *config, int32_t participant_index, char *msg, size_t msgsize)
{
DDSRT_STATIC_ASSERT (DDSI_PORT_MULTI_DISC >= 0 &&
DDSI_PORT_MULTI_DISC + 1 == DDSI_PORT_MULTI_DATA &&
@ -103,7 +103,7 @@ bool ddsi_valid_portmapping (const struct ddsi_portmapping *map, uint32_t domain
int n = snprintf (msg, msgsize, "port number(s) of out range:");
size_t pos = (n >= 0 && (size_t) n <= msgsize) ? (size_t) n : msgsize;
do {
if (!get_port_int (&dummy_port, map, which, domain_id, participant_index, str, sizeof (str)))
if (!get_port_int (&dummy_port, &config->ports, which, config->extDomainId.value, participant_index, str, sizeof (str)))
{
n = snprintf (msg + pos, msgsize - pos, "%s %s %s", ok ? "" : ",", portname (which), str);
if (n >= 0 && (size_t) n <= msgsize - pos)
@ -114,12 +114,12 @@ bool ddsi_valid_portmapping (const struct ddsi_portmapping *map, uint32_t domain
return ok;
}
uint32_t ddsi_get_port (const struct ddsi_portmapping *map, enum ddsi_port which, uint32_t domain_id, int32_t participant_index)
uint32_t ddsi_get_port (const struct config *config, enum ddsi_port which, int32_t participant_index)
{
/* Not supposed to come here if port mapping is invalid */
uint32_t port;
char str[32];
bool ok = get_port_int (&port, map, which, domain_id, participant_index, str, sizeof (str));
bool ok = get_port_int (&port, &config->ports, which, config->extDomainId.value, participant_index, str, sizeof (str));
assert (ok);
(void) ok;
return port;

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

@ -105,7 +105,7 @@ static int add_addresses_to_addrset_1 (const struct q_globals *gv, struct addrse
assert (gv->config.maxAutoParticipantIndex >= 0);
for (int32_t i = 0; i <= gv->config.maxAutoParticipantIndex; i++)
{
loc.port = ddsi_get_port (&gv->config.ports, DDSI_PORT_UNI_DISC, gv->config.domainId, i);
loc.port = ddsi_get_port (&gv->config, DDSI_PORT_UNI_DISC, i);
if (i == 0)
GVLOG (DDS_LC_CONFIG, "%s", ddsi_locator_to_string(gv, buf, sizeof(buf), &loc));
else
@ -116,7 +116,7 @@ static int add_addresses_to_addrset_1 (const struct q_globals *gv, struct addrse
else
{
if (port_mode == -1)
loc.port = ddsi_get_port (&gv->config.ports, DDSI_PORT_MULTI_DISC, gv->config.domainId, 0);
loc.port = ddsi_get_port (&gv->config, DDSI_PORT_MULTI_DISC, 0);
else
loc.port = (uint32_t) port_mode;
GVLOG (DDS_LC_CONFIG, "%s", ddsi_locator_to_string(gv, buf, sizeof(buf), &loc));

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

@ -697,6 +697,10 @@ static const struct cfgelem discovery_peers_cfgelems[] = {
};
static const struct cfgelem discovery_cfgelems[] = {
{ LEAF("Tag"), 0, "", ABSOFF(domainTag), 0, uf_string, ff_free, pf_string,
BLURB("<p>String extension for domain id that remote participants must match to be discovered.</p>") },
{ LEAF ("ExternalDomainId"), 1, "default", ABSOFF (extDomainId), 0, uf_maybe_int32, 0, pf_maybe_int32,
BLURB("<p>An override for the domain id, to be used in discovery and for determining the port number mapping. This allows creating multiple domains in a single process while making them appear as a single domain on the network. The value \"default\" disables the override.</p>") },
{ LEAF("DSGracePeriod"), 1, "30 s", ABSOFF(ds_grace_period), 0, uf_duration_inf, 0, pf_duration,
BLURB("<p>This setting controls for how long endpoints discovered via a Cloud discovery service will survive after the discovery service disappeared, allowing reconnect without loss of data when the discovery service restarts (or another instance takes over).</p>") },
{ GROUP("Peers", discovery_peers_cfgelems),

27
src/core/ddsi/src/q_ddsi_discovery.c
View file

@ -216,12 +216,25 @@ int spdp_write (struct participant *pp)
nn_plist_init_empty (&ps);
ps.present |= PP_PARTICIPANT_GUID | PP_BUILTIN_ENDPOINT_SET |
PP_PROTOCOL_VERSION | PP_VENDORID | PP_PARTICIPANT_LEASE_DURATION;
PP_PROTOCOL_VERSION | PP_VENDORID | PP_PARTICIPANT_LEASE_DURATION |
PP_DOMAIN_ID;
ps.participant_guid = pp->e.guid;
ps.builtin_endpoint_set = pp->bes;
ps.protocol_version.major = RTPS_MAJOR;
ps.protocol_version.minor = RTPS_MINOR;
ps.vendorid = NN_VENDORID_ECLIPSE;
ps.domain_id = pp->e.gv->config.extDomainId.value;
/* Be sure not to send a DOMAIN_TAG when it is the default (an empty)
string: it is an "incompatible-if-unrecognized" parameter, and so
implementations that don't understand the parameter will refuse to
discover us, and so sending the default would break backwards
compatibility. */
if (strcmp (pp->e.gv->config.domainTag, "") != 0)
{
ps.present |= PP_DOMAIN_TAG;
ps.aliased |= PP_DOMAIN_TAG;
ps.domain_tag = pp->e.gv->config.domainTag;
}
if (pp->prismtech_bes)
{
ps.present |= PP_PRISMTECH_BUILTIN_ENDPOINT_SET;
@ -520,6 +533,18 @@ static int handle_SPDP_alive (const struct receiver_state *rst, seqno_t seq, nn_
dds_duration_t lease_duration;
unsigned custom_flags = 0;
/* If advertised domain id or domain tag doesn't match, ignore the message. Do this first to
minimize the impact such messages have. */
{
const uint32_t domain_id = (datap->present & PP_DOMAIN_ID) ? datap->domain_id : gv->config.extDomainId.value;
const char *domain_tag = (datap->present & PP_DOMAIN_TAG) ? datap->domain_tag : "";
if (domain_id != gv->config.extDomainId.value || strcmp (domain_tag, gv->config.domainTag) != 0)
{
GVTRACE ("ignore remote participant in mismatching domain %"PRIu32" tag \"%s\"\n", domain_id, domain_tag);
return 0;
}
}
if (!(datap->present & PP_PARTICIPANT_GUID) || !(datap->present & PP_BUILTIN_ENDPOINT_SET))
{
GVWARNING ("data (SPDP, vendor %u.%u): no/invalid payload\n", rst->vendor.id[0], rst->vendor.id[1]);

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

@ -82,7 +82,7 @@ static enum make_uc_sockets_ret make_uc_sockets (struct q_globals *gv, uint32_t
if (gv->config.many_sockets_mode == MSM_NO_UNICAST)
{
assert (ppid == PARTICIPANT_INDEX_NONE);
*pdata = *pdisc = ddsi_get_port (&gv->config.ports, DDSI_PORT_MULTI_DISC, gv->config.domainId, ppid);
*pdata = *pdisc = ddsi_get_port (&gv->config, DDSI_PORT_MULTI_DISC, ppid);
if (gv->config.allowMulticast)
{
/* FIXME: ugly hack - but we'll fix up after creating the multicast sockets */
@ -90,8 +90,8 @@ static enum make_uc_sockets_ret make_uc_sockets (struct q_globals *gv, uint32_t
}
}
*pdisc = ddsi_get_port (&gv->config.ports, DDSI_PORT_UNI_DISC, gv->config.domainId, ppid);
*pdata = ddsi_get_port (&gv->config.ports, DDSI_PORT_UNI_DATA, gv->config.domainId, ppid);
*pdisc = ddsi_get_port (&gv->config, DDSI_PORT_UNI_DISC, ppid);
*pdata = ddsi_get_port (&gv->config, DDSI_PORT_UNI_DATA, ppid);
if (!ddsi_is_valid_port (gv->m_factory, *pdisc) || !ddsi_is_valid_port (gv->m_factory, *pdata))
return MUSRET_INVALID_PORTS;
@ -279,7 +279,7 @@ static int string_to_default_locator (const struct q_globals *gv, nn_locator_t *
static int set_spdp_address (struct q_globals *gv)
{
const uint32_t port = ddsi_get_port (&gv->config.ports, DDSI_PORT_MULTI_DISC, gv->config.domainId, 0);
const uint32_t port = ddsi_get_port (&gv->config, DDSI_PORT_MULTI_DISC, 0);
int rc = 0;
/* FIXME: FIXME: FIXME: */
gv->loc_spdp_mc.kind = NN_LOCATOR_KIND_INVALID;
@ -311,7 +311,7 @@ static int set_spdp_address (struct q_globals *gv)
static int set_default_mc_address (struct q_globals *gv)
{
const uint32_t port = ddsi_get_port (&gv->config.ports, DDSI_PORT_MULTI_DATA, gv->config.domainId, 0);
const uint32_t port = ddsi_get_port (&gv->config, DDSI_PORT_MULTI_DATA, 0);
int rc;
if (!gv->config.defaultMulticastAddressString)
gv->loc_default_mc = gv->loc_spdp_mc;
@ -452,6 +452,14 @@ int rtps_config_prep (struct q_globals *gv, struct cfgst *cfgst)
unsigned num_channel_threads = 0;
#endif
/* advertised domain id defaults to the real domain id; clear "isdefault" so the config
dump includes the actually used value rather than "default" */
if (gv->config.extDomainId.isdefault)
{
gv->config.extDomainId.value = gv->config.domainId;
gv->config.extDomainId.isdefault = 0;
}
{
char message[256];
int32_t ppidx;
@ -464,7 +472,7 @@ int rtps_config_prep (struct q_globals *gv, struct cfgst *cfgst)
assert (0);
ppidx = 0;
}
if (!ddsi_valid_portmapping (&gv->config.ports, gv->config.domainId, ppidx, message, sizeof (message)))
if (!ddsi_valid_portmapping (&gv->config, ppidx, message, sizeof (message)))
{
DDS_ILOG (DDS_LC_ERROR, gv->config.domainId, "Invalid port mapping: %s\n", message);
goto err_config_late_error;
@ -658,11 +666,11 @@ int create_multicast_sockets (struct q_globals *gv)
uint32_t port;
qos->m_multicast = 1;
port = ddsi_get_port (&gv->config.ports, DDSI_PORT_MULTI_DISC, gv->config.domainId, 0);
port = ddsi_get_port (&gv->config, DDSI_PORT_MULTI_DISC, 0);
if (!ddsi_is_valid_port (gv->m_factory, port))
{
GVERROR ("Failed to create discovery multicast socket for domain %"PRIu32": resulting port number (%"PRIu32") is out of range\n",
gv->config.domainId, port);
gv->config.extDomainId.value, port);
goto err_disc;
}
if ((disc = ddsi_factory_create_conn (gv->m_factory, port, qos)) == NULL)
@ -674,11 +682,11 @@ int create_multicast_sockets (struct q_globals *gv)
}
else
{
port = ddsi_get_port (&gv->config.ports, DDSI_PORT_MULTI_DATA, gv->config.domainId, 0);
port = ddsi_get_port (&gv->config, DDSI_PORT_MULTI_DATA, 0);
if (!ddsi_is_valid_port (gv->m_factory, port))
{
GVERROR ("Failed to create data multicast socket for domain %"PRIu32": resulting port number (%"PRIu32") is out of range\n",
gv->config.domainId, port);
gv->config.extDomainId.value, port);
goto err_disc;
}
if ((data = ddsi_factory_create_conn (gv->m_factory, port, qos)) == NULL)
@ -1005,7 +1013,7 @@ int rtps_init (struct q_globals *gv)
#ifdef DDSI_INCLUDE_NETWORK_PARTITIONS
/* Convert address sets in partition mappings from string to address sets */
{
const uint32_t port = ddsi_get_port (&gv->config.ports, DDSI_PORT_MULTI_DATA, gv->config.domainId, 0);
const uint32_t port = ddsi_get_port (&gv->config, DDSI_PORT_MULTI_DATA, 0);
struct config_networkpartition_listelem *np;
for (np = gv->config.networkPartitions; np; np = np->next)
{
@ -1102,10 +1110,10 @@ int rtps_init (struct q_globals *gv)
break;
case MUSRET_INVALID_PORTS:
GVERROR ("Failed to create unicast sockets for domain %"PRIu32" participant index %d: resulting port numbers (%"PRIu32", %"PRIu32") are out of range\n",
gv->config.domainId, gv->config.participantIndex, port_disc_uc, port_data_uc);
gv->config.extDomainId.value, gv->config.participantIndex, port_disc_uc, port_data_uc);
goto err_unicast_sockets;
case MUSRET_NOSOCKET:
GVERROR ("rtps_init: failed to create unicast sockets for domain %"PRId32" participant index %d (ports %"PRIu32", %"PRIu32")\n", gv->config.domainId, gv->config.participantIndex, port_disc_uc, port_data_uc);
GVERROR ("rtps_init: failed to create unicast sockets for domain %"PRId32" participant index %d (ports %"PRIu32", %"PRIu32")\n", gv->config.extDomainId.value, gv->config.participantIndex, port_disc_uc, port_data_uc);
goto err_unicast_sockets;
}
}
@ -1124,7 +1132,7 @@ int rtps_init (struct q_globals *gv)
break;
case MUSRET_INVALID_PORTS:
GVERROR ("Failed to create unicast sockets for domain %"PRIu32" participant index %d: resulting port numbers (%"PRIu32", %"PRIu32") are out of range\n",
gv->config.domainId, ppid, port_disc_uc, port_data_uc);
gv->config.extDomainId.value, ppid, port_disc_uc, port_data_uc);
goto err_unicast_sockets;
case MUSRET_NOSOCKET: /* Try next one */
break;
@ -1132,7 +1140,7 @@ int rtps_init (struct q_globals *gv)
}
if (ppid > gv->config.maxAutoParticipantIndex)
{
GVERROR ("Failed to find a free participant index for domain %"PRIu32"\n", gv->config.domainId);
GVERROR ("Failed to find a free participant index for domain %"PRIu32"\n", gv->config.extDomainId.value);
goto err_unicast_sockets;
}
gv->config.participantIndex = ppid;

12
src/core/ddsi/src/q_plist.c
View file

@ -1173,6 +1173,8 @@ static const struct piddesc piddesc_omg[] = {
#ifdef DDSI_INCLUDE_SSM
PPV (READER_FAVOURS_SSM, reader_favours_ssm, Xu),
#endif
PP (DOMAIN_ID, domain_id, Xu),
PP (DOMAIN_TAG, domain_tag, XS),
{ PID_STATUSINFO, PDF_FUNCTION, PP_STATUSINFO, "STATUSINFO",
offsetof (struct nn_plist, statusinfo), membersize (struct nn_plist, statusinfo),
{ .f = { .deser = deser_statusinfo, .ser = ser_statusinfo } }, 0 },
@ -1318,8 +1320,8 @@ static const struct piddesc_index piddesc_vendor_index[] = {
/* List of entries that require unalias, fini processing;
initialized by nn_plist_init_tables; will assert when
table too small or too large */
static const struct piddesc *piddesc_unalias[19];
static const struct piddesc *piddesc_fini[19];
static const struct piddesc *piddesc_unalias[20];
static const struct piddesc *piddesc_fini[20];
static ddsrt_once_t table_init_control = DDSRT_ONCE_INIT;
static nn_parameterid_t pid_without_flags (nn_parameterid_t pid)
@ -2098,13 +2100,7 @@ static dds_return_t init_one_parameter (nn_plist_t *plist, nn_ipaddress_params_t
return return_unrecognized_pid (plist, pid);
assert (pid_without_flags (pid) == pid_without_flags (entry->pid));
if (pid != entry->pid)
{
DDS_CERROR (logcfg, "error processing parameter list (vendor %u.%u, version %u.%u): pid %"PRIx16" mapped to pid %"PRIx16"\n",
dd->vendorid.id[0], dd->vendorid.id[1],
dd->protocol_version.major, dd->protocol_version.minor,
pid, entry->pid);
return return_unrecognized_pid (plist, pid);
}
assert (pid != PID_PAD);
struct flagset flagset;