ignore data until a heartbeat is received (#146)

When data arrives before a heartbeat has been received, it is impossible
to know whether this is a new "live" sample or a retransmit, and for
this reason the requesting of historical data is delayed until a
heartbeat arrives that informs the readers of the range of sequence
numbers to request as historical data.

However, by this time, and without this new condition in place, the
reader may have already received some data directly, and may
consequently request some data twice.  That's not right.

Requiring a heartbeat to have been received before delivering the data
avoids this problem, but potentially delays receiving data after a new
writer/reader pair has been matched.  The delay caused by a full
handshake at that point seems less bad that the odd case of stuttering
where that isn't expected.  There are almost certainly some tricks
possible to avoid that delay in the common cases, but there are more
important things to do ...

Best-effort readers on a reliable proxy writer are a bit special: if
there are only best-effort readers, there is no guarantee that a
heartbeat will be received, and so the condition does not apply.  This
commit attempts to deal with that by only requiring a heartbeat if some
reliable readers exist, but that doesn't allow a smooth transition from
"only best-effort readers" to "some reliable readers".

One could moreover argue that this condition should not be imposed on
volatile readers (at worst you get a little bit of data from before the
match), but equally well that it should (there's no guarantee that no
sample would be skipped in the case of a keep-all writer, if the first
sample happened to be a retransmit).

Signed-off-by: Erik Boasson <eb@ilities.com>

src/core/ddsi/src/q_entity.c

@@ -4111,12 +4111,13 @@ int new_proxy_writer (const struct nn_guid *ppguid, const struct nn_guid *guid,
   } else {
     pwr->deliver_synchronously = 0;
   }
-  pwr->have_seen_heartbeat = 0;
+  /* Pretend we have seen a heartbeat if the proxy writer is a best-effort one */
+  isreliable = (pwr->c.xqos->reliability.kind != NN_BEST_EFFORT_RELIABILITY_QOS);
+  pwr->have_seen_heartbeat = !isreliable;
   pwr->local_matching_inprogress = 1;
 #ifdef DDSI_INCLUDE_SSM
   pwr->supports_ssm = (addrset_contains_ssm (as) && config.allowMulticast & AMC_SSM) ? 1 : 0;
 #endif
-  isreliable = (pwr->c.xqos->reliability.kind != NN_BEST_EFFORT_RELIABILITY_QOS);
 
   /* Only assert PP lease on receipt of data if enabled (duh) and the proxy participant is a
      "real" participant, rather than the thing we use for endpoints discovered via the DS */

src/core/ddsi/src/q_receive.c

@@ -1745,6 +1745,13 @@ static int handle_Gap (struct receiver_state *rst, nn_etime_t tnow, struct nn_rm
   }
   DDS_TRACE("%x:%x:%x:%x -> %x:%x:%x:%x", PGUID (src), PGUID (dst));
 
+  if (!pwr->have_seen_heartbeat && pwr->n_reliable_readers > 0)
+  {
+    DDS_TRACE(": no heartbeat seen yet", PGUID (pwr->e.guid), PGUID (dst));
+    ddsrt_mutex_unlock (&pwr->e.lock);
+    return 1;
+  }
+
   /* Notify reordering in proxy writer & and the addressed reader (if
      it is out-of-sync, &c.), while delivering samples that become
      available because preceding ones are now known to be missing. */
@@ -2173,6 +2180,25 @@ static void handle_regular (struct receiver_state *rst, nn_etime_t tnow, struct
 
   /* Shouldn't lock the full writer, but will do so for now */
   ddsrt_mutex_lock (&pwr->e.lock);
+
+  /* Don't accept data when reliable readers exist and we haven't yet seen
+     a heartbeat telling us what the "current" sequence number of the writer
+     is. If no reliable readers are present, we can't request a heartbeat and
+     therefore must not require one.
+
+     This should be fine except for the one case where one transitions from
+     having only best-effort readers to also having a reliable reader (in
+     the same process): in that case, the requirement that a heartbeat has
+     been seen could potentially result in a disruption of the data flow to
+     the best-effort readers.  That state should last only for a very short
+     time, but it is rather inelegant.  */
+  if (!pwr->have_seen_heartbeat && pwr->n_reliable_readers > 0)
+  {
+    ddsrt_mutex_unlock (&pwr->e.lock);
+    DDS_TRACE(" %x:%x:%x:%x -> %x:%x:%x:%x: no heartbeat seen yet", PGUID (pwr->e.guid), PGUID (dst));
+    return;
+  }
+
   if (ut_avlIsEmpty (&pwr->readers) || pwr->local_matching_inprogress)
   {
     ddsrt_mutex_unlock (&pwr->e.lock);