Fix issue in dds_create_topic_arbitrary (#422)

* Fix issue in dds_create_topic_arbitrary

Changed the behaviour of dds_create_topic_arbitrary with respect to the
sertopic parameter: the existing function dds_create_topic_arbitrary is
marked deprecated and replaced by dds_create_topic_generic, which returns
the sertopic that is actually used in as an out parameter. This can be eiter
the provided sertopic (if this sertopic was not yet known in the domain) or an
existing sertopic if the sertopic was registered earlier.

Signed-off-by: Dennis Potman <dennis.potman@adlinktech.com>

* Fix memory leaks in case topic creation fails.

Signed-off-by: Dennis Potman <dennis.potman@adlinktech.com>

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

@@ -998,25 +998,29 @@ dds_create_topic(
   const dds_qos_t *qos,
   const dds_listener_t *listener);
 
+
+#define DDS_HAS_CREATE_TOPIC_GENERIC 1
 /**
- * @brief Creates a new topic with arbitrary type handling.
+ * @brief Creates a new topic with provided type handling.
  *
  * The type name for the topic is taken from the provided "sertopic" object. Topic
  * matching is done on a combination of topic name and type name. Each successful
  * call to dds_create_topic creates a new topic entity sharing the same QoS
  * settings with all other topics of the same name.
  *
- * If sertopic is not yet known in the domain, it is added and its refcount
- * incremented; if an equivalent sertopic object is already known, then the known
- * one is used instead.
+ * In case this function returns a valid handle, the ownership of the provided
+ * sertopic is handed over to Cyclone. On return, the caller gets in the sertopic parameter a
+ * pointer to the sertopic that is actually used by the topic. This can be the provided sertopic
+ * (if this sertopic was not yet known in the domain), or a sertopic thas was
+ * already known in the domain.
  *
  * @param[in]     participant  Participant on which to create the topic.
- * @param[in]  sertopic     Internal description of the topic type (includes name).
+ * @param[in,out] sertopic     Internal description of the topic type (includes name). On return, the sertopic parameter is set to the actual sertopic that is used by the topic.
  * @param[in]     qos          QoS to set on the new topic (can be NULL).
  * @param[in]     listener     Any listener functions associated with the new topic (can be NULL).
  * @param[in]     sedp_plist   Topic description to be published as part of discovery (if NULL, not published).
  *
- * @returns A valid, unique topic handle or an error code.
+ * @returns A valid, unique topic handle or an error code. Iff a valid handle, the domain takes ownership of provided serdata.
  *
  * @retval >=0
  *             A valid unique topic handle.
@@ -1031,6 +1035,14 @@ dds_create_topic(
  *             topic's type name.
  */
 DDS_EXPORT dds_entity_t
+dds_create_topic_generic (
+  dds_entity_t participant,
+  struct ddsi_sertopic **sertopic,
+  const dds_qos_t *qos,
+  const dds_listener_t *listener,
+  const struct ddsi_plist *sedp_plist);
+
+DDS_DEPRECATED_EXPORT dds_entity_t
 dds_create_topic_arbitrary (
   dds_entity_t participant,
   struct ddsi_sertopic *sertopic,

src/core/ddsc/src/dds_builtin.c

@@ -76,7 +76,12 @@ dds_entity_t dds__get_builtin_topic (dds_entity_t entity, dds_entity_t topic)
   }
 
   dds_qos_t *qos = dds__create_builtin_qos ();
-  tp = dds_create_topic_arbitrary (par->m_entity.m_hdllink.hdl, sertopic, qos, NULL, NULL);
+  if ((tp = dds_create_topic_generic (par->m_entity.m_hdllink.hdl, &sertopic, qos, NULL, NULL)) > 0)
+  {
+    /* keep ownership for built-in sertopics because there are re-used, lifetime for these
+       sertopics is bound to domain */
+    ddsi_sertopic_ref (sertopic);
+  }
   dds_delete_qos (qos);
   dds_entity_unpin (e);
   return tp;

src/core/ddsc/src/dds_topic.c

@@ -276,7 +276,7 @@ static dds_entity_t create_topic_pp_locked (struct dds_participant *pp, struct d
   return hdl;
 }
 
-dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_sertopic *sertopic, const dds_qos_t *qos, const dds_listener_t *listener, const ddsi_plist_t *sedp_plist)
+dds_entity_t dds_create_topic_generic (dds_entity_t participant, struct ddsi_sertopic **sertopic, const dds_qos_t *qos, const dds_listener_t *listener, const ddsi_plist_t *sedp_plist)
 {
   dds_return_t rc;
   dds_participant *pp;
@@ -284,7 +284,7 @@ dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_s
   dds_entity_t hdl;
   struct ddsi_sertopic *sertopic_registered;
 
-  if (sertopic == NULL)
+  if (sertopic == NULL || *sertopic == NULL)
     return DDS_RETCODE_BAD_PARAMETER;
 
   {
@@ -323,14 +323,14 @@ dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_s
 
   /* See if we're allowed to create the topic; ktp is returned pinned & locked
      so we can be sure it doesn't disappear and its QoS can't change */
-  GVTRACE ("dds_create_topic_arbitrary (pp %p "PGUIDFMT" sertopic %p reg?%s refc %"PRIu32" %s/%s)\n",
-           (void *) pp, PGUID (pp->m_entity.m_guid), (void *) sertopic, sertopic->gv ? "yes" : "no",
-           ddsrt_atomic_ld32 (&sertopic->refc), sertopic->name, sertopic->type_name);
+  GVTRACE ("dds_create_topic_generic (pp %p "PGUIDFMT" sertopic %p reg?%s refc %"PRIu32" %s/%s)\n",
+           (void *) pp, PGUID (pp->m_entity.m_guid), (void *) (*sertopic), (*sertopic)->gv ? "yes" : "no",
+           ddsrt_atomic_ld32 (&(*sertopic)->refc), (*sertopic)->name, (*sertopic)->type_name);
   ddsrt_mutex_lock (&pp->m_entity.m_mutex);
   struct dds_ktopic *ktp;
-  if ((rc = lookup_and_check_ktopic (&ktp, pp, sertopic->name, sertopic->type_name, new_qos)) != DDS_RETCODE_OK)
+  if ((rc = lookup_and_check_ktopic (&ktp, pp, (*sertopic)->name, (*sertopic)->type_name, new_qos)) != DDS_RETCODE_OK)
   {
-    GVTRACE ("dds_create_topic_arbitrary: failed after compatibility check: %s\n", dds_strretcode (rc));
+    GVTRACE ("dds_create_topic_generic: failed after compatibility check: %s\n", dds_strretcode (rc));
     dds_participant_unlock (pp);
     dds_delete_qos (new_qos);
     return rc;
@@ -345,8 +345,8 @@ dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_s
     ktp->defer_set_qos = 0;
     ktp->qos = new_qos;
     /* have to copy these because the ktopic can outlast any specific sertopic */
-    ktp->name = ddsrt_strdup (sertopic->name);
-    ktp->type_name = ddsrt_strdup (sertopic->type_name);
+    ktp->name = ddsrt_strdup ((*sertopic)->name);
+    ktp->type_name = ddsrt_strdup ((*sertopic)->type_name);
     ddsrt_avl_insert (&participant_ktopics_treedef, &pp->m_ktopics, ktp);
     GVTRACE ("create_and_lock_ktopic: ktp %p\n", (void *) ktp);
   }
@@ -359,13 +359,13 @@ dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_s
   /* Sertopic: re-use a previously registered one if possible, else register this one */
   {
     ddsrt_mutex_lock (&gv->sertopics_lock);
-    if ((sertopic_registered = ddsi_sertopic_lookup_locked (gv, sertopic)) != NULL)
-      GVTRACE ("dds_create_topic_arbitrary: reuse sertopic %p\n", (void *) sertopic_registered);
+    if ((sertopic_registered = ddsi_sertopic_lookup_locked (gv, *sertopic)) != NULL)
+      GVTRACE ("dds_create_topic_generic: reuse sertopic %p\n", (void *) sertopic_registered);
     else
     {
-      GVTRACE ("dds_create_topic_arbitrary: register new sertopic %p\n", (void *) sertopic);
-      ddsi_sertopic_register_locked (gv, sertopic);
-      sertopic_registered = sertopic;
+      GVTRACE ("dds_create_topic_generic: register new sertopic %p\n", (void *) (*sertopic));
+      ddsi_sertopic_register_locked (gv, *sertopic);
+      sertopic_registered = *sertopic;
     }
     ddsrt_mutex_unlock (&gv->sertopics_lock);
   }
@@ -373,14 +373,27 @@ dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_s
   /* Create topic referencing ktopic & sertopic_registered */
   /* FIXME: setting "implicit" based on sertopic->ops is a hack */
   hdl = create_topic_pp_locked (pp, ktp, (sertopic_registered->ops == &ddsi_sertopic_ops_builtintopic), sertopic_registered, listener, sedp_plist);
+  ddsi_sertopic_unref (*sertopic);
+  *sertopic = sertopic_registered;
   dds_participant_unlock (pp);
-  GVTRACE ("dds_create_topic_arbitrary: new topic %"PRId32"\n", hdl);
+  GVTRACE ("dds_create_topic_generic: new topic %"PRId32"\n", hdl);
   return hdl;
 }
 
+dds_entity_t dds_create_topic_arbitrary (dds_entity_t participant, struct ddsi_sertopic *sertopic, const dds_qos_t *qos, const dds_listener_t *listener, const ddsi_plist_t *sedp_plist)
+{
+  dds_entity_t ret;
+  struct ddsi_sertopic *st = sertopic;
+  ddsi_sertopic_ref (st);
+  if ((ret = dds_create_topic_generic (participant, &st, qos, listener, sedp_plist)) < 0)
+    ddsi_sertopic_unref (st);
+  return ret;
+}
+
 dds_entity_t dds_create_topic (dds_entity_t participant, const dds_topic_descriptor_t *desc, const char *name, const dds_qos_t *qos, const dds_listener_t *listener)
 {
   struct ddsi_sertopic_default *st;
+  struct ddsi_sertopic *st_tmp;
   ddsi_plist_t plist;
   dds_entity_t hdl;
   struct dds_entity *ppent;
@@ -433,8 +446,10 @@ dds_entity_t dds_create_topic (dds_entity_t participant, const dds_topic_descrip
       plist.qos.subscription_keys.key_list.strs[index] = dds_string_dup (desc->m_keys[index].m_name);
   }
 
-  hdl = dds_create_topic_arbitrary (participant, &st->c, qos, listener, &plist);
-  ddsi_sertopic_unref (&st->c);
+  st_tmp = &st->c;
+  hdl = dds_create_topic_generic (participant, &st_tmp, qos, listener, &plist);
+  if (hdl < 0)
+    ddsi_sertopic_unref (st_tmp);
   dds_entity_unpin (ppent);
   ddsi_plist_fini (&plist);
   return hdl;

src/core/ddsi/src/ddsi_sertopic.c

@@ -74,7 +74,6 @@ struct ddsi_sertopic *ddsi_sertopic_lookup_locked (struct ddsi_domaingv *gv, con
 void ddsi_sertopic_register_locked (struct ddsi_domaingv *gv, struct ddsi_sertopic *sertopic)
 {
   assert (sertopic->gv == NULL);
-  assert (ddsrt_atomic_ld32 (&sertopic->refc) == 1);
 
   (void) ddsi_sertopic_ref (sertopic);
   sertopic->gv = gv;