auto-dispose, reconnects, readcdr, static analysis, refactoring

docs/manual/options.md

@@ -1349,7 +1349,7 @@ is therefore recommended to set it to at least several seconds.
 Valid values are finite durations with an explicit unit or the keyword
 'inf' for infinity. Recognised units: ns, us, ms, s, min, hr, day.
 
-The default value is: "10s".
+The default value is: "0s".
 
 
 #### //CycloneDDS/Domain/Internal/RediscoveryBlacklistDuration[@enforce]

etc/cyclonedds.rnc

@@ -1113,7 +1113,7 @@ is therefore recommended to set it to at least several seconds.</p>
 
 <p>Valid values are finite durations with an explicit unit or the keyword
 'inf' for infinity. Recognised units: ns, us, ms, s, min, hr,
-day.</p><p>The default value is: &quot;10s&quot;.</p>""" ] ]
+day.</p><p>The default value is: &quot;0s&quot;.</p>""" ] ]
         element RediscoveryBlacklistDuration {
           [ a:documentation [ xml:lang="en" """
 <p>This attribute controls whether the configured time during which

etc/cyclonedds.xsd

@@ -1475,7 +1475,7 @@ is therefore recommended to set it to at least several seconds.</p>
 
 <p>Valid values are finite durations with an explicit unit or the keyword
 'inf' for infinity. Recognised units: ns, us, ms, s, min, hr,
-day.&lt;/p&gt;&lt;p&gt;The default value is: &amp;quot;10s&amp;quot;.&lt;/p&gt;</xs:documentation>
+day.&lt;/p&gt;&lt;p&gt;The default value is: &amp;quot;0s&amp;quot;.&lt;/p&gt;</xs:documentation>
     </xs:annotation>
     <xs:complexType>
       <xs:simpleContent>

src/core/ddsc/include/dds/dds.h

@@ -90,6 +90,7 @@ typedef enum dds_status_id {
   DDS_PUBLICATION_MATCHED_STATUS_ID,
   DDS_SUBSCRIPTION_MATCHED_STATUS_ID
 } dds_status_id_t;
+#define DDS_STATUS_ID_MAX (DDS_SUBSCRIPTION_MATCHED_STATUS_ID)
 
 /** Another topic exists with the same name but with different characteristics. */
 #define DDS_INCONSISTENT_TOPIC_STATUS          (1u << DDS_INCONSISTENT_TOPIC_STATUS_ID)
@@ -2756,6 +2757,53 @@ dds_take_mask_wl(
   uint32_t maxs,
   uint32_t mask);
 
+#define DDS_HAS_READCDR 1
+/**
+ * @brief Access the collection of serialized data values (of same type) and
+ *        sample info from the data reader, readcondition or querycondition.
+ *
+ * This call accesses the serialized data from the data reader, readcondition or
+ * querycondition and makes it available to the application. The serialized data
+ * is made available through \ref ddsi_serdata structures. Returned samples are
+ * marked as READ.
+ *
+ * Return value provides information about the number of samples read, which will
+ * be <= maxs. Based on the count, the buffer will contain serialized data to be
+ * read only when valid_data bit in sample info structure is set.
+ * The buffer required for data values, could be allocated explicitly or can
+ * use the memory from data reader to prevent copy. In the latter case, buffer and
+ * sample_info should be returned back, once it is no longer using the data.
+ *
+ * @param[in]  reader_or_condition Reader, readcondition or querycondition entity.
+ * @param[out] buf An array of pointers to \ref ddsi_serdata structures that contain
+ *                 the serialized data. The pointers can be NULL.
+ * @param[in]  maxs Maximum number of samples to read.
+ * @param[out] si Pointer to an array of \ref dds_sample_info_t returned for each data value.
+ * @param[in]  mask Filter the data based on dds_sample_state_t|dds_view_state_t|dds_instance_state_t.
+ *
+ * @returns A dds_return_t with the number of samples read or an error code.
+ *
+ * @retval >=0
+ *             Number of samples read.
+ * @retval DDS_RETCODE_ERROR
+ *             An internal error has occurred.
+ * @retval DDS_RETCODE_BAD_PARAMETER
+ *             One of the given arguments is not valid.
+ * @retval DDS_RETCODE_ILLEGAL_OPERATION
+ *             The operation is invoked on an inappropriate object.
+ * @retval DDS_RETCODE_ALREADY_DELETED
+ *             The entity has already been deleted.
+ * @retval DDS_RETCODE_PRECONDITION_NOT_MET
+ *             The precondition for this operation is not met.
+ */
+DDS_EXPORT dds_return_t
+dds_readcdr(
+  dds_entity_t reader_or_condition,
+  struct ddsi_serdata **buf,
+  uint32_t maxs,
+  dds_sample_info_t *si,
+  uint32_t mask);
+
 /**
  * @brief Access the collection of serialized data values (of same type) and
  *        sample info from the data reader, readcondition or querycondition.

src/core/ddsc/include/dds/ddsc/dds_public_impl.h

@@ -225,6 +225,13 @@ enum dds_stream_typecode_subtype {
 #define DDS_OP_FLAG_KEY 0x01 /* key field: applicable to {1,2,4,8}BY, STR, BST, ARR-of-{1,2,4,8}BY */
 #define DDS_OP_FLAG_DEF 0x02 /* union has a default case (for DDS_OP_ADR | DDS_OP_TYPE_UNI) */
 
+/* For a union: (1) the discriminator may be a key field; (2) there may be a default value;
+   and (3) the discriminator can be an integral type (or enumerated - here treated as equivalent).
+   What it can't be is a floating-point type. So DEF and FP need never be set at the same time.
+   There are only a few flag bits, so saving one is not such a bad idea. */
+#define DDS_OP_FLAG_FP  0x02 /* floating-point: applicable to {4,8}BY and arrays, sequences of them */
+#define DDS_OP_FLAG_SGN 0x04 /* signed: applicable to {1,2,4,8}BY and arrays, sequences of them */
+
 /**
  * Description : Enable or disable write batching. Overrides default configuration
  * setting for write batching (Internal/WriteBatch).

src/core/ddsc/include/dds/ddsc/dds_rhc.h

@@ -27,9 +27,8 @@ struct dds_reader;
 struct ddsi_tkmap;
 
 typedef dds_return_t (*dds_rhc_associate_t) (struct dds_rhc *rhc, struct dds_reader *reader, const struct ddsi_sertopic *topic, struct ddsi_tkmap *tkmap);
-typedef int (*dds_rhc_read_t) (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond);
-typedef int (*dds_rhc_take_t) (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond);
-typedef int (*dds_rhc_takecdr_t) (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle);
+typedef int32_t (*dds_rhc_read_take_t) (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond);
+typedef int32_t (*dds_rhc_read_take_cdr_t) (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle);
 
 typedef bool (*dds_rhc_add_readcondition_t) (struct dds_rhc *rhc, struct dds_readcond *cond);
 typedef void (*dds_rhc_remove_readcondition_t) (struct dds_rhc *rhc, struct dds_readcond *cond);
@@ -40,9 +39,10 @@ struct dds_rhc_ops {
   /* A copy of DDSI rhc ops comes first so we can use either interface without
      additional indirections */
   struct ddsi_rhc_ops rhc_ops;
-  dds_rhc_read_t read;
-  dds_rhc_take_t take;
-  dds_rhc_takecdr_t takecdr;
+  dds_rhc_read_take_t read;
+  dds_rhc_read_take_t take;
+  dds_rhc_read_take_cdr_t readcdr;
+  dds_rhc_read_take_cdr_t takecdr;
   dds_rhc_add_readcondition_t add_readcondition;
   dds_rhc_remove_readcondition_t remove_readcondition;
   dds_rhc_lock_samples_t lock_samples;
@@ -76,13 +76,16 @@ DDS_EXPORT inline void dds_rhc_set_qos (struct dds_rhc *rhc, const struct dds_qo
 DDS_EXPORT inline void dds_rhc_free (struct dds_rhc *rhc) {
   rhc->common.ops->rhc_ops.free (&rhc->common.rhc);
 }
-DDS_EXPORT inline int dds_rhc_read (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond) {
+DDS_EXPORT inline int32_t dds_rhc_read (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond) {
   return (rhc->common.ops->read) (rhc, lock, values, info_seq, max_samples, mask, handle, cond);
 }
-DDS_EXPORT inline int dds_rhc_take (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond) {
+DDS_EXPORT inline int32_t dds_rhc_take (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond) {
   return rhc->common.ops->take (rhc, lock, values, info_seq, max_samples, mask, handle, cond);
 }
-DDS_EXPORT inline int dds_rhc_takecdr (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle) {
+DDS_EXPORT inline int32_t dds_rhc_readcdr (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle) {
+  return rhc->common.ops->readcdr (rhc, lock, values, info_seq, max_samples, sample_states, view_states, instance_states, handle);
+}
+DDS_EXPORT inline int32_t dds_rhc_takecdr (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle) {
   return rhc->common.ops->takecdr (rhc, lock, values, info_seq, max_samples, sample_states, view_states, instance_states, handle);
 }
 DDS_EXPORT inline bool dds_rhc_add_readcondition (struct dds_rhc *rhc, struct dds_readcond *cond) {

src/core/ddsc/src/dds__writer.h

@@ -23,7 +23,7 @@ DEFINE_ENTITY_LOCK_UNLOCK(inline, dds_writer, DDS_KIND_WRITER)
 struct status_cb_data;
 
 void dds_writer_status_cb (void *entity, const struct status_cb_data * data);
-dds_return_t dds__writer_wait_for_acks (struct dds_writer *wr, dds_time_t abstimeout);
+DDS_EXPORT dds_return_t dds__writer_wait_for_acks (struct dds_writer *wr, ddsi_guid_t *rdguid, dds_time_t abstimeout);
 
 #if defined (__cplusplus)
 }

src/core/ddsc/src/dds_builtin.c

@@ -15,6 +15,7 @@
 #include "dds/ddsi/q_entity.h"
 #include "dds/ddsi/q_thread.h"
 #include "dds/ddsi/q_config.h"
+#include "dds/ddsi/q_bswap.h"
 #include "dds/ddsi/ddsi_domaingv.h"
 #include "dds/ddsi/ddsi_plist.h"
 #include "dds__init.h"
@@ -181,10 +182,13 @@ static struct ddsi_tkmap_instance *dds__builtin_get_tkmap_entry (const struct dd
   struct dds_domain *domain = vdomain;
   struct ddsi_tkmap_instance *tk;
   struct ddsi_serdata *sd;
-  struct ddsi_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 (domain->builtin_participant_topic, &kh);
+  union { ddsi_guid_t guid; struct ddsi_keyhash keyhash; } x;
+  x.guid = nn_hton_guid (*guid);
+  /* 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 (domain->builtin_participant_topic, &x.keyhash);
   tk = ddsi_tkmap_find (domain->gv.m_tkmap, sd, true);
   ddsi_serdata_unref (sd);
   return tk;
@@ -196,7 +200,7 @@ struct ddsi_serdata *dds__builtin_make_sample (const struct entity_common *e, dd
   struct dds_domain *dom = e->gv->builtin_topic_interface->arg;
   struct ddsi_sertopic *topic = NULL;
   struct ddsi_serdata *serdata;
-  struct ddsi_keyhash keyhash;
+  union { ddsi_guid_t guid; struct ddsi_keyhash keyhash; } x;
   switch (e->kind)
   {
     case EK_PARTICIPANT:
@@ -213,8 +217,8 @@ struct ddsi_serdata *dds__builtin_make_sample (const struct entity_common *e, dd
       break;
   }
   assert (topic != NULL);
-  memcpy (&keyhash, &e->guid, sizeof (keyhash));
-  serdata = ddsi_serdata_from_keyhash (topic, &keyhash);
+  x.guid = nn_hton_guid (e->guid);
+  serdata = ddsi_serdata_from_keyhash (topic, &x.keyhash);
   serdata->timestamp = timestamp;
   serdata->statusinfo = alive ? 0 : NN_STATUSINFO_DISPOSE | NN_STATUSINFO_UNREGISTER;
   return serdata;

src/core/ddsc/src/dds_listener.c

@@ -90,8 +90,7 @@ void dds_listener_copy(dds_listener_t * __restrict dst, const dds_listener_t * _
 static bool dds_combine_listener_merge (uint32_t inherited, void (*dst)(void), void (*src)(void))
 {
   (void)inherited;
-  (void)src;
-  return dst == 0;
+  return dst == 0 && src != 0;
 }
 
 static bool dds_combine_listener_override_inherited (uint32_t inherited, void (*dst)(void), void (*src)(void))

src/core/ddsc/src/dds_matched.c

@@ -144,8 +144,7 @@ static dds_builtintopic_endpoint_t *make_builtintopic_endpoint (const ddsi_guid_
 dds_builtintopic_endpoint_t *dds_get_matched_subscription_data (dds_entity_t writer, dds_instance_handle_t ih)
 {
   dds_writer *wr;
-  dds_return_t rc;
-  if ((rc = dds_writer_lock (writer, &wr)) != DDS_RETCODE_OK)
+  if (dds_writer_lock (writer, &wr))
     return NULL;
   else
   {
@@ -187,8 +186,7 @@ dds_builtintopic_endpoint_t *dds_get_matched_subscription_data (dds_entity_t wri
 dds_builtintopic_endpoint_t *dds_get_matched_publication_data (dds_entity_t reader, dds_instance_handle_t ih)
 {
   dds_reader *rd;
-  dds_return_t rc;
-  if ((rc = dds_reader_lock (reader, &rd)) != DDS_RETCODE_OK)
+  if (dds_reader_lock (reader, &rd))
     return NULL;
   else
   {

src/core/ddsc/src/dds_publisher.c

@@ -114,7 +114,7 @@ dds_return_t dds_wait_for_acks (dds_entity_t publisher_or_writer, dds_duration_t
       return DDS_RETCODE_UNSUPPORTED;
 
     case DDS_KIND_WRITER:
-      ret = dds__writer_wait_for_acks ((struct dds_writer *) p_or_w_ent, abstimeout);
+      ret = dds__writer_wait_for_acks ((struct dds_writer *) p_or_w_ent, NULL, abstimeout);
       dds_entity_unpin (p_or_w_ent);
       return ret;
 

src/core/ddsc/src/dds_read.c

@@ -40,7 +40,7 @@ static dds_return_t dds_read_impl (bool take, dds_entity_t reader_or_condition,
 #define NC_FREE_BUF 2u
 #define NC_RESET_BUF 4u
 
-  if (buf == NULL || si == NULL || maxs == 0 || bufsz == 0 || bufsz < maxs)
+  if (buf == NULL || si == NULL || maxs == 0 || bufsz == 0 || bufsz < maxs || maxs > INT32_MAX)
     return DDS_RETCODE_BAD_PARAMETER;
 
   if ((ret = dds_entity_pin (reader_or_condition, &entity)) < 0) {
@@ -61,14 +61,6 @@ static dds_return_t dds_read_impl (bool take, dds_entity_t reader_or_condition,
 
   thread_state_awake (ts1, &entity->m_domain->gv);
 
-  if (hand != DDS_HANDLE_NIL)
-  {
-    if (ddsi_tkmap_find_by_id (entity->m_domain->gv.m_tkmap, hand) == NULL) {
-      ret = DDS_RETCODE_PRECONDITION_NOT_MET;
-      goto fail_awake_pinned;
-    }
-  }
-
   /* Allocate samples if not provided (assuming all or none provided) */
   if (buf[0] == NULL)
   {
@@ -142,8 +134,6 @@ static dds_return_t dds_read_impl (bool take, dds_entity_t reader_or_condition,
 #undef NC_FREE_BUF
 #undef NC_RESET_BUF
 
-fail_awake_pinned:
-  thread_state_asleep (ts1);
 fail_pinned:
   dds_entity_unpin (entity);
 fail:
@@ -157,12 +147,8 @@ static dds_return_t dds_readcdr_impl (bool take, dds_entity_t reader_or_conditio
   struct dds_reader *rd;
   struct dds_entity *entity;
 
-  assert (take);
-  assert (buf);
-  assert (si);
-  assert (hand == DDS_HANDLE_NIL);
-  assert (maxs > 0);
-  (void)take;
+  if (buf == NULL || si == NULL || maxs == 0 || maxs > INT32_MAX)
+    return DDS_RETCODE_BAD_PARAMETER;
 
   if ((ret = dds_entity_pin (reader_or_condition, &entity)) < 0) {
     return ret;
@@ -185,7 +171,11 @@ static dds_return_t dds_readcdr_impl (bool take, dds_entity_t reader_or_conditio
   assert (dds_entity_kind (rd->m_entity.m_parent) == DDS_KIND_SUBSCRIBER);
   dds_entity_status_reset (rd->m_entity.m_parent, DDS_DATA_ON_READERS_STATUS);
 
-  ret = dds_rhc_takecdr (rd->m_rhc, lock, buf, si, maxs, mask & DDS_ANY_SAMPLE_STATE, mask & DDS_ANY_VIEW_STATE, mask & DDS_ANY_INSTANCE_STATE, hand);
+  if (take)
+    ret = dds_rhc_takecdr (rd->m_rhc, lock, buf, si, maxs, mask & DDS_ANY_SAMPLE_STATE, mask & DDS_ANY_VIEW_STATE, mask & DDS_ANY_INSTANCE_STATE, hand);
+  else
+    ret = dds_rhc_readcdr (rd->m_rhc, lock, buf, si, maxs, mask & DDS_ANY_SAMPLE_STATE, mask & DDS_ANY_VIEW_STATE, mask & DDS_ANY_INSTANCE_STATE, hand);
+
   dds_entity_unpin (entity);
   thread_state_asleep (ts1);
   return ret;
@@ -239,6 +229,18 @@ dds_return_t dds_read_mask_wl (dds_entity_t rd_or_cnd, void **buf, dds_sample_in
   return dds_read_impl (false, rd_or_cnd, buf, maxs, maxs, si, mask, DDS_HANDLE_NIL, lock, false);
 }
 
+dds_return_t dds_readcdr (dds_entity_t rd_or_cnd, struct ddsi_serdata **buf, uint32_t maxs, dds_sample_info_t *si, uint32_t mask)
+{
+  bool lock = true;
+  if (maxs == DDS_READ_WITHOUT_LOCK)
+  {
+    lock = false;
+    /* FIXME: Fix the interface. */
+    maxs = 100;
+  }
+  return dds_readcdr_impl (false, rd_or_cnd, buf, maxs, si, mask, DDS_HANDLE_NIL, lock);
+}
+
 dds_return_t dds_read_instance (dds_entity_t rd_or_cnd, void **buf, dds_sample_info_t *si, size_t bufsz, uint32_t maxs, dds_instance_handle_t handle)
 {
   bool lock = true;

src/core/ddsc/src/dds_rhc.c

@@ -22,6 +22,7 @@ extern inline void dds_rhc_set_qos (struct dds_rhc *rhc, const struct dds_qos *q
 extern inline void dds_rhc_free (struct dds_rhc *rhc);
 extern inline int dds_rhc_read (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond);
 extern inline int dds_rhc_take (struct dds_rhc *rhc, bool lock, void **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t mask, dds_instance_handle_t handle, struct dds_readcond *cond);
+extern inline int dds_rhc_readcdr (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle);
 extern inline int dds_rhc_takecdr (struct dds_rhc *rhc, bool lock, struct ddsi_serdata **values, dds_sample_info_t *info_seq, uint32_t max_samples, uint32_t sample_states, uint32_t view_states, uint32_t instance_states, dds_instance_handle_t handle);
 extern inline bool dds_rhc_add_readcondition (struct dds_rhc *rhc, struct dds_readcond *cond);
 extern inline void dds_rhc_remove_readcondition (struct dds_rhc *rhc, struct dds_readcond *cond);