Potential fix for the race condition

Applied suggestions from code review

Author: urmahp

include/ur_client_library/comm/producer.h

@@ -43,7 +43,7 @@ class URProducer : public IProducer<T>
   URStream<T>& stream_;
   Parser<T>& parser_;
   std::chrono::seconds timeout_;
-  std::function<void()> on_rtde_reconnect_cb_;
+  std::function<void()> on_reconnect_cb_;
 
   bool running_;
 
@@ -133,17 +133,10 @@ class URProducer : public IProducer<T>
       if (stream_.closed())
         return false;
 
-      if (stream_.getStreamType() == URStreamType::RTDE)
+      if (on_reconnect_cb_)
       {
-        if (on_rtde_reconnect_cb_)
-        {
-          URCL_LOG_WARN("Failed to read from RTDE stream, invoking on reconnect callback and stopping the producer");
-          on_rtde_reconnect_cb_();
-        }
-        else
-        {
-          URCL_LOG_ERROR("Failed to read from RTDE stream without a reconnect handler stopping the producer");
-        }
+        URCL_LOG_WARN("Failed to read from stream, invoking on reconnect callback and stopping the producer");
+        on_reconnect_cb_();
         return false;
       }
 
@@ -162,14 +155,15 @@ class URProducer : public IProducer<T>
   }
 
   /*!
-   * \brief Sets the RTDE reconnection callback. RTDE requires setting up the communication again upon reconnection
-   * it is not enough to just reconnect to the stream.
+   * \brief Sets the reconnection callback. Use this to configure a reconnection callback instead of connecting directly
+   * to the stream again. This is needed for RTDE as it requires setting up the communication again upon reconnection it
+   * is not enough to just reconnect to the stream.
    *
-   * \param on_rtde_reconnect_cb Callback to be invoked when connection is lost to the RTDE stream.
+   * \param on_reconnect_cb Callback to be invoked when connection is lost to the stream.
    */
-  void setRTDEReconnectionCallback(std::function<void()> on_rtde_reconnect_cb)
+  void setReconnectionCallback(std::function<void()> on_reconnect_cb)
   {
-    on_rtde_reconnect_cb_ = on_rtde_reconnect_cb;
+    on_reconnect_cb_ = on_reconnect_cb;
   }
 };
 }  // namespace comm

include/ur_client_library/comm/stream.h

@@ -31,17 +31,6 @@ namespace urcl
 namespace comm
 {
 
-/*!
- * \brief Different types of UR streams
- */
-enum class URStreamType
-{
-  Primary = 30001,    ///< Stream connected to the primary interface
-  Secondary = 30002,  ///< Stream connected to the secondary interface
-  RTDE = 30004,       ///< Stream connected to the RTDE interface
-  UNKNOWN = -1,       ///< Stream type is fetched from the port, this is to handle unknown ports
-};
-
 /*!
  * \brief The stream is an abstraction of the TCPSocket that offers reading a full UR data package
  * out of the socket. This means, it has to have some knowledge about the package structure to
@@ -129,26 +118,6 @@ class URStream : public TCPSocket
     return host_;
   }
 
-  /*!
-   * \brief Get the stream type
-   *
-   * \returns The stream type
-   */
-  URStreamType getStreamType()
-  {
-    switch (port_)
-    {
-      case static_cast<int>(URStreamType::Primary):
-        return URStreamType::Primary;
-      case static_cast<int>(URStreamType::Secondary):
-        return URStreamType::Secondary;
-      case static_cast<int>(URStreamType::RTDE):
-        return URStreamType::RTDE;
-      default:
-        return URStreamType::UNKNOWN;
-    }
-  }
-
 private:
   std::string host_;
   int port_;

include/ur_client_library/rtde/rtde_client.h

@@ -232,6 +232,7 @@ class RTDEClient
   std::unique_ptr<comm::Pipeline<RTDEPackage>> pipeline_;
   RTDEWriter writer_;
   bool reconnecting_;
+  bool stop_reconnection_;
   std::mutex reconnect_mutex_;
   std::thread reconnecting_thread_;
 

src/rtde/rtde_client.cpp

@@ -47,6 +47,7 @@ RTDEClient::RTDEClient(std::string robot_ip, comm::INotifier& notifier, const st
   , pipeline_(std::make_unique<comm::Pipeline<RTDEPackage>>(*prod_, PIPELINE_NAME, notifier, true))
   , writer_(&stream_, input_recipe_)
   , reconnecting_(false)
+  , stop_reconnection_(false)
   , max_frequency_(URE_MAX_FREQUENCY)
   , target_frequency_(target_frequency)
   , client_state_(ClientState::UNINITIALIZED)
@@ -71,6 +72,7 @@ RTDEClient::RTDEClient(std::string robot_ip, comm::INotifier& notifier, const st
   , pipeline_(std::make_unique<comm::Pipeline<RTDEPackage>>(*prod_, PIPELINE_NAME, notifier, true))
   , writer_(&stream_, input_recipe_)
   , reconnecting_(false)
+  , stop_reconnection_(false)
   , max_frequency_(URE_MAX_FREQUENCY)
   , target_frequency_(target_frequency)
   , client_state_(ClientState::UNINITIALIZED)
@@ -79,7 +81,8 @@ RTDEClient::RTDEClient(std::string robot_ip, comm::INotifier& notifier, const st
 
 RTDEClient::~RTDEClient()
 {
-  reconnecting_ = false;
+  prod_->setReconnectionCallback(nullptr);
+  stop_reconnection_ = true;
   if (reconnecting_thread_.joinable())
   {
     reconnecting_thread_.join();
@@ -100,7 +103,7 @@ bool RTDEClient::init(const size_t max_connection_attempts, const std::chrono::m
     return true;
   }
 
-  prod_->setRTDEReconnectionCallback(std::bind(&RTDEClient::reconnectCallback, this));
+  prod_->setReconnectionCallback(nullptr);
 
   unsigned int attempts = 0;
   std::stringstream ss;
@@ -121,6 +124,9 @@ bool RTDEClient::init(const size_t max_connection_attempts, const std::chrono::m
   // Stop pipeline again
   pipeline_->stop();
   client_state_ = ClientState::INITIALIZED;
+  // Set reconnection callback after we are initialized to ensure that a disconnect during initialization doesn't
+  // trigger a reconnect
+  prod_->setReconnectionCallback(std::bind(&RTDEClient::reconnectCallback, this));
   return true;
 }
 
@@ -296,7 +302,6 @@ void RTDEClient::resetOutputRecipe(const std::vector<std::string> new_recipe)
 
   parser_ = RTDEParser(output_recipe_);
   prod_ = std::make_unique<comm::URProducer<RTDEPackage>>(stream_, parser_);
-  prod_->setRTDEReconnectionCallback(std::bind(&RTDEClient::reconnectCallback, this));
   pipeline_ = std::make_unique<comm::Pipeline<RTDEPackage>>(*prod_, PIPELINE_NAME, notifier_, true);
 }
 
@@ -611,7 +616,6 @@ bool RTDEClient::sendStart()
       ss << "Did not receive answer to RTDE start request. Message received instead: " << std::endl
          << package->toString();
       URCL_LOG_WARN("%s", ss.str().c_str());
-      return false;
     }
   }
   std::stringstream ss;
@@ -644,7 +648,6 @@ bool RTDEClient::sendPause()
     }
     if (rtde_interface::ControlPackagePause* tmp = dynamic_cast<rtde_interface::ControlPackagePause*>(package.get()))
     {
-      client_state_ = ClientState::PAUSED;
       return tmp->accepted_;
     }
   }
@@ -774,7 +777,7 @@ void RTDEClient::reconnect()
     }
 
     const std::string reconnecting_stopped_msg = "Reconnecting has been stopped, because the object is being destroyed";
-    if (reconnecting_ == false)
+    if (stop_reconnection_)
     {
       URCL_LOG_WARN(reconnecting_stopped_msg.c_str());
       return;
@@ -806,7 +809,7 @@ void RTDEClient::reconnect()
     while (std::chrono::steady_clock::now() - start_time < duration)
     {
       std::this_thread::sleep_for(std::chrono::milliseconds(250));
-      if (reconnecting_ == false)
+      if (stop_reconnection_)
       {
         URCL_LOG_WARN(reconnecting_stopped_msg.c_str());
         return;
@@ -828,12 +831,13 @@ void RTDEClient::reconnect()
   {
     pause();
   }
+  URCL_LOG_INFO("Done reconnecting to the RTDE interface");
   reconnecting_ = false;
 }
 
 void RTDEClient::reconnectCallback()
 {
-  if (reconnecting_)
+  if (reconnecting_ || stop_reconnection_)
   {
     return;
   }