Fix topic creation race condition

rmw_connext_cpp/src/rmw_publisher.cpp

@@ -20,6 +20,7 @@
 #include "rmw/rmw.h"
 #include "rmw/types.h"
 
+#include "rmw_connext_shared_cpp/create_topic.hpp"
 #include "rmw_connext_shared_cpp/qos.hpp"
 #include "rmw_connext_shared_cpp/types.hpp"
 
@@ -122,7 +123,6 @@ rmw_create_publisher(
   DDS::Publisher * dds_publisher = nullptr;
   DDS::DataWriter * topic_writer = nullptr;
   DDS::Topic * topic = nullptr;
-  DDS::TopicDescription * topic_description = nullptr;
   void * info_buf = nullptr;
   void * listener_buf = nullptr;
   ConnextPublisherListener * publisher_listener = nullptr;
@@ -192,34 +192,7 @@ rmw_create_publisher(
     goto fail;
   }
 
-  topic_description = participant->lookup_topicdescription(topic_str);
-  if (!topic_description) {
-    DDS::TopicQos default_topic_qos;
-    status = participant->get_default_topic_qos(default_topic_qos);
-    if (status != DDS::RETCODE_OK) {
-      RMW_SET_ERROR_MSG("failed to get default topic qos");
-      goto fail;
-    }
-
-    topic = participant->create_topic(
-      topic_str, type_name.c_str(),
-      default_topic_qos, NULL, DDS::STATUS_MASK_NONE);
-    if (!topic) {
-      RMW_SET_ERROR_MSG_WITH_FORMAT_STRING(
-        "failed to create topic '%s' for node namespace='%s' name='%s'",
-        topic_name, node->namespace_, node->name);
-      goto fail;
-    }
-  } else {
-    DDS::Duration_t timeout = DDS::Duration_t::from_seconds(0);
-    topic = participant->find_topic(topic_str, timeout);
-    if (!topic) {
-      RMW_SET_ERROR_MSG_WITH_FORMAT_STRING(
-        "failed to find topic '%s' for node namespace='%s' name='%s'",
-        topic_name, node->namespace_, node->name);
-      goto fail;
-    }
-  }
+  topic = rmw_connext_shared_cpp::create_topic(node, topic_name, topic_str, type_name.c_str());
   DDS::String_free(topic_str);
   topic_str = nullptr;
 

rmw_connext_cpp/src/rmw_subscription.cpp

@@ -19,6 +19,7 @@
 #include "rmw/impl/cpp/macros.hpp"
 #include "rmw/rmw.h"
 
+#include "rmw_connext_shared_cpp/create_topic.hpp"
 #include "rmw_connext_shared_cpp/qos.hpp"
 #include "rmw_connext_shared_cpp/types.hpp"
 
@@ -115,7 +116,6 @@ rmw_create_subscription(
   DDS::ReturnCode_t status;
   DDS::Subscriber * dds_subscriber = nullptr;
   DDS::Topic * topic = nullptr;
-  DDS::TopicDescription * topic_description = nullptr;
   DDS::DataReader * topic_reader = nullptr;
   DDS::ReadCondition * read_condition = nullptr;
   void * info_buf = nullptr;
@@ -186,34 +186,7 @@ rmw_create_subscription(
     goto fail;
   }
 
-  topic_description = participant->lookup_topicdescription(topic_str);
-  if (!topic_description) {
-    DDS::TopicQos default_topic_qos;
-    status = participant->get_default_topic_qos(default_topic_qos);
-    if (status != DDS::RETCODE_OK) {
-      RMW_SET_ERROR_MSG("failed to get default topic qos");
-      goto fail;
-    }
-
-    topic = participant->create_topic(
-      topic_str, type_name.c_str(),
-      default_topic_qos, NULL, DDS::STATUS_MASK_NONE);
-    if (!topic) {
-      RMW_SET_ERROR_MSG_WITH_FORMAT_STRING(
-        "failed to create topic '%s' for node namespace='%s' name='%s'",
-        topic_name, node->namespace_, node->name);
-      goto fail;
-    }
-  } else {
-    DDS::Duration_t timeout = DDS::Duration_t::from_seconds(0);
-    topic = participant->find_topic(topic_str, timeout);
-    if (!topic) {
-      RMW_SET_ERROR_MSG_WITH_FORMAT_STRING(
-        "failed to find topic '%s' for node namespace='%s' name='%s'",
-        topic_name, node->namespace_, node->name);
-      goto fail;
-    }
-  }
+  topic = rmw_connext_shared_cpp::create_topic(node, topic_name, topic_str, type_name.c_str());
   DDS::String_free(topic_str);
   topic_str = nullptr;
 

rmw_connext_shared_cpp/CMakeLists.txt

@@ -38,6 +38,7 @@ add_library(
   SHARED
   src/condition_error.cpp
   src/count.cpp
+  src/create_topic.cpp
   src/demangle.cpp
   src/event.cpp
   src/event_converter.cpp

rmw_connext_shared_cpp/include/rmw_connext_shared_cpp/create_topic.hpp

@@ -0,0 +1,35 @@
+// Copyright 2020 Open Source Robotics Foundation, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef RMW_CONNEXT_SHARED_CPP__CREATE_TOPIC_HPP_
+#define RMW_CONNEXT_SHARED_CPP__CREATE_TOPIC_HPP_
+
+#include "rmw/types.h"
+#include "rmw_connext_shared_cpp/ndds_include.hpp"
+#include "rmw_connext_shared_cpp/visibility_control.h"
+
+namespace rmw_connext_shared_cpp
+{
+
+RMW_CONNEXT_SHARED_CPP_PUBLIC
+DDS::Topic *
+create_topic(
+  const rmw_node_t * node,
+  const char * topic_name,
+  const char * dds_topic_name,
+  const char * dds_topic_type);
+
+}  // namespace rmw_connext_shared_cpp
+
+#endif  // RMW_CONNEXT_SHARED_CPP__CREATE_TOPIC_HPP_

rmw_connext_shared_cpp/include/rmw_connext_shared_cpp/types.hpp

@@ -149,6 +149,7 @@ struct ConnextNodeInfo
   CustomPublisherListener * publisher_listener;
   CustomSubscriberListener * subscriber_listener;
   rmw_guard_condition_t * graph_guard_condition;
+  std::mutex topic_creation_mutex;
 };
 
 struct ConnextPublisherGID

rmw_connext_shared_cpp/src/create_topic.cpp

@@ -0,0 +1,76 @@
+// Copyright 2020 Open Source Robotics Foundation, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "rmw_connext_shared_cpp/create_topic.hpp"
+
+#include <cassert>
+#include <mutex>
+
+#include "rmw/error_handling.h"
+#include "rmw/types.h"
+
+#include "rmw_connext_shared_cpp/types.hpp"
+#include "rmw_connext_shared_cpp/ndds_include.hpp"
+
+DDS::Topic *
+rmw_connext_shared_cpp::create_topic(
+  const rmw_node_t * node,
+  const char * topic_name,
+  const char * dds_topic_name,
+  const char * dds_topic_type)
+{
+  // This function is internal, and should be called from functions that already verified
+  // the node validity.
+  assert(node);
+  assert(node->data);
+
+  auto node_info = static_cast<ConnextNodeInfo *>(node->data);
+  auto participant = static_cast<DDS::DomainParticipant *>(node_info->participant);
+
+  std::lock_guard<std::mutex> guard(node_info->topic_creation_mutex);
+
+  DDS::TopicDescription * topic_description =
+    participant->lookup_topicdescription(dds_topic_name);
+
+  DDS::Topic * topic = nullptr;
+  DDS::ReturnCode_t status = DDS::RETCODE_ERROR;
+  if (!topic_description) {
+    DDS::TopicQos default_topic_qos;
+    status = participant->get_default_topic_qos(default_topic_qos);
+    if (status != DDS::RETCODE_OK) {
+      RMW_SET_ERROR_MSG("failed to get default topic qos");
+      return nullptr;
+    }
+
+    topic = participant->create_topic(
+      dds_topic_name, dds_topic_type,
+      default_topic_qos, nullptr, DDS::STATUS_MASK_NONE);
+    if (!topic) {
+      RMW_SET_ERROR_MSG_WITH_FORMAT_STRING(
+        "failed to create topic '%s' for node namespace='%s' name='%s'",
+        topic_name, node->namespace_, node->name);
+      return nullptr;
+    }
+  } else {
+    DDS::Duration_t timeout = DDS::Duration_t::from_seconds(0);
+    topic = participant->find_topic(dds_topic_name, timeout);
+    if (!topic) {
+      RMW_SET_ERROR_MSG_WITH_FORMAT_STRING(
+        "failed to find topic '%s' for node namespace='%s' name='%s'",
+        topic_name, node->namespace_, node->name);
+      return nullptr;
+    }
+  }
+  return topic;
+}