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

Fixed issue that Cyclone does not receive multicast data on Windows when

Browse source 
the destination cache of the network stack is in a certain state. The issue
is resolved by binding unicast sockets (incoming unicast and all outgoing
traffic) to the address of the interface instead of inaddr_any (0.0.0.0).
Set the new configuration option internal/BindUnicastToInterfaceAddr to
false to get the old behavior.

Co-authored-by: Erik Boasson <eb@ilities.com>

Signed-off-by: Dennis Potman <dennis.potman@adlinktech.com>
This commit is contained in:
Dennis Potman 2020-02-20 14:53:54 +01:00 • committed by eboasson
parent af3604dea7
commit 754eb4090e
7 changed files with 54 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

11
docs/manual/options.md
View file

@ -554,7 +554,7 @@ The default value is: "default".
### //CycloneDDS/Domain/Internal
Children: [AccelerateRexmitBlockSize](#cycloneddsdomaininternalacceleraterexmitblocksize), [AssumeMulticastCapable](#cycloneddsdomaininternalassumemulticastcapable), [AutoReschedNackDelay](#cycloneddsdomaininternalautoreschednackdelay), [BuiltinEndpointSet](#cycloneddsdomaininternalbuiltinendpointset), [ControlTopic](#cycloneddsdomaininternalcontroltopic), [DDSI2DirectMaxThreads](#cycloneddsdomaininternalddsi2directmaxthreads), [DefragReliableMaxSamples](#cycloneddsdomaininternaldefragreliablemaxsamples), [DefragUnreliableMaxSamples](#cycloneddsdomaininternaldefragunreliablemaxsamples), [DeliveryQueueMaxSamples](#cycloneddsdomaininternaldeliveryqueuemaxsamples), [EnableExpensiveChecks](#cycloneddsdomaininternalenableexpensivechecks), [GenerateKeyhash](#cycloneddsdomaininternalgeneratekeyhash), [HeartbeatInterval](#cycloneddsdomaininternalheartbeatinterval), [LateAckMode](#cycloneddsdomaininternallateackmode), [LeaseDuration](#cycloneddsdomaininternalleaseduration), [LivelinessMonitoring](#cycloneddsdomaininternallivelinessmonitoring), [MaxParticipants](#cycloneddsdomaininternalmaxparticipants), [MaxQueuedRexmitBytes](#cycloneddsdomaininternalmaxqueuedrexmitbytes), [MaxQueuedRexmitMessages](#cycloneddsdomaininternalmaxqueuedrexmitmessages), [MaxSampleSize](#cycloneddsdomaininternalmaxsamplesize), [MeasureHbToAckLatency](#cycloneddsdomaininternalmeasurehbtoacklatency), [MinimumSocketReceiveBufferSize](#cycloneddsdomaininternalminimumsocketreceivebuffersize), [MinimumSocketSendBufferSize](#cycloneddsdomaininternalminimumsocketsendbuffersize), [MonitorPort](#cycloneddsdomaininternalmonitorport), [MultipleReceiveThreads](#cycloneddsdomaininternalmultiplereceivethreads), [NackDelay](#cycloneddsdomaininternalnackdelay), [PreEmptiveAckDelay](#cycloneddsdomaininternalpreemptiveackdelay), [PrimaryReorderMaxSamples](#cycloneddsdomaininternalprimaryreordermaxsamples), [PrioritizeRetransmit](#cycloneddsdomaininternalprioritizeretransmit), [RediscoveryBlacklistDuration](#cycloneddsdomaininternalrediscoveryblacklistduration), [RetransmitMerging](#cycloneddsdomaininternalretransmitmerging), [RetransmitMergingPeriod](#cycloneddsdomaininternalretransmitmergingperiod), [RetryOnRejectBestEffort](#cycloneddsdomaininternalretryonrejectbesteffort), [SPDPResponseMaxDelay](#cycloneddsdomaininternalspdpresponsemaxdelay), [ScheduleTimeRounding](#cycloneddsdomaininternalscheduletimerounding), [SecondaryReorderMaxSamples](#cycloneddsdomaininternalsecondaryreordermaxsamples), [SendAsync](#cycloneddsdomaininternalsendasync), [SquashParticipants](#cycloneddsdomaininternalsquashparticipants), [SynchronousDeliveryLatencyBound](#cycloneddsdomaininternalsynchronousdeliverylatencybound), [SynchronousDeliveryPriorityThreshold](#cycloneddsdomaininternalsynchronousdeliveryprioritythreshold), [Test](#cycloneddsdomaininternaltest), [UnicastResponseToSPDPMessages](#cycloneddsdomaininternalunicastresponsetospdpmessages), [UseMulticastIfMreqn](#cycloneddsdomaininternalusemulticastifmreqn), [Watermarks](#cycloneddsdomaininternalwatermarks), [WriteBatch](#cycloneddsdomaininternalwritebatch), [WriterLingerDuration](#cycloneddsdomaininternalwriterlingerduration)
Children: [AccelerateRexmitBlockSize](#cycloneddsdomaininternalacceleraterexmitblocksize), [AssumeMulticastCapable](#cycloneddsdomaininternalassumemulticastcapable), [AutoReschedNackDelay](#cycloneddsdomaininternalautoreschednackdelay), [BindUnicastToInterfaceAddr](#cycloneddsdomaininternalbindunicasttointerfaceaddr), [BuiltinEndpointSet](#cycloneddsdomaininternalbuiltinendpointset), [ControlTopic](#cycloneddsdomaininternalcontroltopic), [DDSI2DirectMaxThreads](#cycloneddsdomaininternalddsi2directmaxthreads), [DefragReliableMaxSamples](#cycloneddsdomaininternaldefragreliablemaxsamples), [DefragUnreliableMaxSamples](#cycloneddsdomaininternaldefragunreliablemaxsamples), [DeliveryQueueMaxSamples](#cycloneddsdomaininternaldeliveryqueuemaxsamples), [EnableExpensiveChecks](#cycloneddsdomaininternalenableexpensivechecks), [GenerateKeyhash](#cycloneddsdomaininternalgeneratekeyhash), [HeartbeatInterval](#cycloneddsdomaininternalheartbeatinterval), [LateAckMode](#cycloneddsdomaininternallateackmode), [LeaseDuration](#cycloneddsdomaininternalleaseduration), [LivelinessMonitoring](#cycloneddsdomaininternallivelinessmonitoring), [MaxParticipants](#cycloneddsdomaininternalmaxparticipants), [MaxQueuedRexmitBytes](#cycloneddsdomaininternalmaxqueuedrexmitbytes), [MaxQueuedRexmitMessages](#cycloneddsdomaininternalmaxqueuedrexmitmessages), [MaxSampleSize](#cycloneddsdomaininternalmaxsamplesize), [MeasureHbToAckLatency](#cycloneddsdomaininternalmeasurehbtoacklatency), [MinimumSocketReceiveBufferSize](#cycloneddsdomaininternalminimumsocketreceivebuffersize), [MinimumSocketSendBufferSize](#cycloneddsdomaininternalminimumsocketsendbuffersize), [MonitorPort](#cycloneddsdomaininternalmonitorport), [MultipleReceiveThreads](#cycloneddsdomaininternalmultiplereceivethreads), [NackDelay](#cycloneddsdomaininternalnackdelay), [PreEmptiveAckDelay](#cycloneddsdomaininternalpreemptiveackdelay), [PrimaryReorderMaxSamples](#cycloneddsdomaininternalprimaryreordermaxsamples), [PrioritizeRetransmit](#cycloneddsdomaininternalprioritizeretransmit), [RediscoveryBlacklistDuration](#cycloneddsdomaininternalrediscoveryblacklistduration), [RetransmitMerging](#cycloneddsdomaininternalretransmitmerging), [RetransmitMergingPeriod](#cycloneddsdomaininternalretransmitmergingperiod), [RetryOnRejectBestEffort](#cycloneddsdomaininternalretryonrejectbesteffort), [SPDPResponseMaxDelay](#cycloneddsdomaininternalspdpresponsemaxdelay), [ScheduleTimeRounding](#cycloneddsdomaininternalscheduletimerounding), [SecondaryReorderMaxSamples](#cycloneddsdomaininternalsecondaryreordermaxsamples), [SendAsync](#cycloneddsdomaininternalsendasync), [SquashParticipants](#cycloneddsdomaininternalsquashparticipants), [SynchronousDeliveryLatencyBound](#cycloneddsdomaininternalsynchronousdeliverylatencybound), [SynchronousDeliveryPriorityThreshold](#cycloneddsdomaininternalsynchronousdeliveryprioritythreshold), [Test](#cycloneddsdomaininternaltest), [UnicastResponseToSPDPMessages](#cycloneddsdomaininternalunicastresponsetospdpmessages), [UseMulticastIfMreqn](#cycloneddsdomaininternalusemulticastifmreqn), [Watermarks](#cycloneddsdomaininternalwatermarks), [WriteBatch](#cycloneddsdomaininternalwritebatch), [WriterLingerDuration](#cycloneddsdomaininternalwriterlingerduration)
The Internal elements deal with a variety of settings that evolving and
@ -600,6 +600,15 @@ Valid values are finite durations with an explicit unit or the keyword
The default value is: "1 s".
#### //CycloneDDS/Domain/Internal/BindUnicastToInterfaceAddr
Boolean
Bind unicast sockets to the address of the preferred interface; if false,
bind to 0.0.0.0 (IPv4) or its equivalent
The default value is: "true".
#### //CycloneDDS/Domain/Internal/BuiltinEndpointSet
One of: full, writers, minimal

7
etc/cyclonedds.rnc
View file

@ -491,6 +491,13 @@ day.</p><p>The default value is: &quot;1 s&quot;.</p>""" ] ]
duration_inf
}?
& [ a:documentation [ xml:lang="en" """
<p>Bind unicast sockets to the address of the preferred interface; if
false, bind to 0.0.0.0 (IPv4) or its equivalent</p><p>The default value
is: &quot;true&quot;.</p>""" ] ]
element BindUnicastToInterfaceAddr {
xsd:boolean
}?
& [ a:documentation [ xml:lang="en" """
<p>This element controls which participants will have which built-in
endpoints for the discovery and liveliness protocols. Valid values
are:</p>

9
etc/cyclonedds.xsd
View file

@ -638,6 +638,7 @@ reserved. This includes renaming or moving options.&lt;/p&gt;</xs:documentation>
<xs:element minOccurs="0" ref="config:AccelerateRexmitBlockSize"/>
<xs:element minOccurs="0" ref="config:AssumeMulticastCapable"/>
<xs:element minOccurs="0" ref="config:AutoReschedNackDelay"/>
<xs:element minOccurs="0" ref="config:BindUnicastToInterfaceAddr"/>
<xs:element minOccurs="0" ref="config:BuiltinEndpointSet"/>
<xs:element minOccurs="0" ref="config:ControlTopic"/>
<xs:element minOccurs="0" ref="config:DDSI2DirectMaxThreads"/>
@ -716,6 +717,14 @@ of HEARTBEAT messages.&lt;/p&gt;
day.&lt;/p&gt;&lt;p&gt;The default value is: &amp;quot;1 s&amp;quot;.&lt;/p&gt;</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="BindUnicastToInterfaceAddr" type="xs:boolean">
<xs:annotation>
<xs:documentation>
&lt;p&gt;Bind unicast sockets to the address of the preferred interface; if
false, bind to 0.0.0.0 (IPv4) or its equivalent&lt;/p&gt;&lt;p&gt;The default value
is: &amp;quot;true&amp;quot;.&lt;/p&gt;</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="BuiltinEndpointSet">
<xs:annotation>
<xs:documentation>

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

@ -332,6 +332,7 @@ struct config
int64_t initial_deaf_mute_reset;
int use_multicast_if_mreqn;
int bind_unicast_to_interface_addr;
struct prune_deleted_ppant prune_deleted_ppant;
};

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

@ -35,7 +35,7 @@ struct nn_interface {
char *name;
};
int make_socket (ddsrt_socket_t *socket, uint16_t port, bool stream, bool reuse, const struct ddsi_domaingv *gv);
int make_socket (ddsrt_socket_t *socket, uint16_t port, bool stream, bool multicast, const struct ddsi_domaingv *gv);
int find_own_ip (struct ddsi_domaingv *gv, const char *requested_address);
uint32_t locator_to_hopefully_unique_uint32 (const nn_locator_t *src);

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

@ -578,6 +578,8 @@ static const struct cfgelem unsupp_cfgelems[] = {
BLURB("<p>This element controls whether retransmits are prioritized over new data, speeding up recovery.</p>") },
{ LEAF("UseMulticastIfMreqn"), 1, "0", ABSOFF(use_multicast_if_mreqn), 0, uf_int, 0, pf_int,
BLURB("<p>Do not use.</p>") },
{ LEAF("BindUnicastToInterfaceAddr"), 1, "true", ABSOFF(bind_unicast_to_interface_addr), 0, uf_boolean, 0, pf_boolean,
BLURB("<p>Bind unicast sockets to the address of the preferred interface; if false, bind to 0.0.0.0 (IPv4) or its equivalent</p>") },
{ LEAF("SendAsync"), 1, "false", ABSOFF(xpack_send_async), 0, uf_boolean, 0, pf_boolean,
BLURB("<p>This element controls whether the actual sending of packets occurs on the same thread that prepares them, or is done asynchronously by another thread.</p>") },
{ LEAF_W_ATTRS("RediscoveryBlacklistDuration", rediscovery_blacklist_duration_attrs), 1, "10s", ABSOFF(prune_deleted_ppant.delay), 0, uf_duration_inf, 0, pf_duration,

41
src/core/ddsi/src/q_nwif.c
View file

@ -214,32 +214,39 @@ static int set_reuse_options (const struct ddsrt_log_cfg *logcfg, ddsrt_socket_t
return 0;
}
static int bind_socket (ddsrt_socket_t socket, unsigned short port, const struct ddsi_domaingv *gv)
static int bind_socket (ddsrt_socket_t socket, unsigned short port, bool multicast, const struct ddsi_domaingv *gv)
{
dds_return_t rc = DDS_RETCODE_ERROR;
#if DDSRT_HAVE_IPV6
if (gv->config.transport_selector == TRANS_TCP6 || gv->config.transport_selector == TRANS_UDP6)
{
struct sockaddr_in6 socketname;
memset (&socketname, 0, sizeof (socketname));
socketname.sin6_family = AF_INET6;
socketname.sin6_port = htons (port);
socketname.sin6_addr = ddsrt_in6addr_any;
if (IN6_IS_ADDR_LINKLOCAL (&socketname.sin6_addr)) {
socketname.sin6_scope_id = gv->interfaceNo;
union {
struct sockaddr_storage x;
struct sockaddr_in6 a;
} socketname;
ddsi_ipaddr_from_loc (&socketname.x, &gv->ownloc);
if (multicast || !gv->config.bind_unicast_to_interface_addr)
socketname.a.sin6_addr = ddsrt_in6addr_any;
socketname.a.sin6_port = htons (port);
if (IN6_IS_ADDR_LINKLOCAL (&socketname.a.sin6_addr)) {
socketname.a.sin6_scope_id = gv->interfaceNo;
}
rc = ddsrt_bind (socket, (struct sockaddr *) &socketname, sizeof (socketname));
rc = ddsrt_bind (socket, (struct sockaddr *) &socketname.a, sizeof (socketname.a));
}
else
#endif
if (gv->config.transport_selector == TRANS_TCP || gv->config.transport_selector == TRANS_UDP)
{
struct sockaddr_in socketname;
socketname.sin_family = AF_INET;
socketname.sin_port = htons (port);
socketname.sin_addr.s_addr = htonl (INADDR_ANY);
rc = ddsrt_bind (socket, (struct sockaddr *) &socketname, sizeof (socketname));
union {
struct sockaddr_storage x;
struct sockaddr_in a;
} socketname;
ddsi_ipaddr_from_loc (&socketname.x, &gv->ownloc);
if (multicast || !gv->config.bind_unicast_to_interface_addr)
socketname.a.sin_addr.s_addr = htonl (INADDR_ANY);
socketname.a.sin_port = htons (port);
rc = ddsrt_bind (socket, (struct sockaddr *) &socketname.a, sizeof (socketname.a));
}
if (rc != DDS_RETCODE_OK && rc != DDS_RETCODE_PRECONDITION_NOT_MET)
{
@ -338,7 +345,7 @@ static int set_mc_options_transmit (ddsrt_socket_t socket, const struct ddsi_dom
}
}
int make_socket (ddsrt_socket_t *sock, uint16_t port, bool stream, bool reuse, const struct ddsi_domaingv *gv)
int make_socket (ddsrt_socket_t *sock, uint16_t port, bool stream, bool multicast, const struct ddsi_domaingv *gv)
{
/* FIXME: this stuff has to move to the transports */
int rc = -2;
@ -366,7 +373,7 @@ int make_socket (ddsrt_socket_t *sock, uint16_t port, bool stream, bool reuse, c
return rc;
}
if (port && reuse && ((rc = set_reuse_options (&gv->logconfig, *sock)) < 0))
if (port && multicast && ((rc = set_reuse_options (&gv->logconfig, *sock)) < 0))
{
goto fail;
}
@ -376,7 +383,7 @@ int make_socket (ddsrt_socket_t *sock, uint16_t port, bool stream, bool reuse, c
(rc = set_rcvbuf (&gv->logconfig, *sock, &gv->config.socket_min_rcvbuf_size) < 0) ||
(rc = set_sndbuf (&gv->logconfig, *sock, gv->config.socket_min_sndbuf_size) < 0) ||
((rc = maybe_set_dont_route (&gv->logconfig, *sock, &gv->config)) < 0) ||
((rc = bind_socket (*sock, port, gv)) < 0)
((rc = bind_socket (*sock, port, multicast, gv)) < 0)
)
{
goto fail;