[Core] Fix Race Conditions in CMsgSubscriber (#1324)

- Properly implement destructor (in inherited classes)! -> Destroy may not be called in base CMsgSubscriber class as this can lead to stack unwinding (most likely causes an exception in destructor, or leads to already deleted functions of derived class being called when they are already deleted) - Correct order of RemReceiveCallback and resetting internal pointer - protect function pointer with mutex. - Make testcase "harder" to more easily trigger race conditions.
eclipse-ecal · Jan 25, 2024 · b5666d5 · b5666d5
1 parent 353529a
commit b5666d5
Show file tree

Hide file tree

Showing 7 changed files with 104 additions and 52 deletions.
diff --git a/ecal/core/include/ecal/msg/capnproto/subscriber.h b/ecal/core/include/ecal/msg/capnproto/subscriber.h
@@ -67,6 +67,14 @@ namespace eCAL
       {
       }
 
+      /**
+       * @brief  Destructor
+      **/
+      ~CBuilderSubscriber() override
+      {
+        this->Destroy();
+      }
+
       /**
       * @brief  Copy Constructor is not available.
       **/

diff --git a/ecal/core/include/ecal/msg/flatbuffers/subscriber.h b/ecal/core/include/ecal/msg/flatbuffers/subscriber.h
@@ -56,6 +56,14 @@ namespace eCAL
       {
       }
 
+      /**
+       * @brief  Destructor
+      **/
+      ~CSubscriber() override
+      {
+        this->Destroy();
+      }
+
       /**
       * @brief  Copy Constructor is not available.
       **/

diff --git a/ecal/core/include/ecal/msg/messagepack/subscriber.h b/ecal/core/include/ecal/msg/messagepack/subscriber.h
@@ -59,6 +59,14 @@ namespace eCAL
       {
       }
 
+      /**
+       * @brief  Destructor
+      **/
+      ~CSubscriber() override
+      {
+        this->Destroy();
+      }
+
       /**
       * @brief  Copy Constructor is not available.
       **/

diff --git a/ecal/core/include/ecal/msg/protobuf/subscriber.h b/ecal/core/include/ecal/msg/protobuf/subscriber.h
@@ -75,6 +75,14 @@ namespace eCAL
       {
       }
 
+      /**
+       * @brief  Destructor
+      **/
+      ~CSubscriber() override
+      {
+        this->Destroy();
+      }
+
       /**
       * @brief  Copy Constructor is not available.
       **/

diff --git a/ecal/core/include/ecal/msg/string/subscriber.h b/ecal/core/include/ecal/msg/string/subscriber.h
@@ -62,6 +62,14 @@ namespace eCAL
       {
       }
 
+      /**
+       * @brief  Destructor
+      **/
+      ~CSubscriber() override
+      {
+        this->Destroy();
+      }
+
       /**
       * @brief  Copy Constructor is not available.
       **/

diff --git a/ecal/core/include/ecal/msg/subscriber.h b/ecal/core/include/ecal/msg/subscriber.h
@@ -64,6 +64,8 @@ namespace eCAL
     {
     }
 
+    virtual ~CMsgSubscriber() = default;
+
     /**
     * @brief  Copy Constructor is not available.
     **/
@@ -113,8 +115,6 @@ namespace eCAL
       return *this;
     }
 
-    virtual ~CMsgSubscriber() {}
-
     /**
      * @brief Creates this object. 
      *
@@ -181,7 +181,10 @@ namespace eCAL
       assert(IsCreated());
       RemReceiveCallback();
 
-      m_cb_callback = callback_;
+      {
+        std::lock_guard<std::mutex> callback_lock(m_cb_callback_mutex);
+        m_cb_callback = callback_;
+      }
       auto callback = std::bind(&CMsgSubscriber::ReceiveCallback, this, std::placeholders::_1, std::placeholders::_2);
       return(CSubscriber::AddReceiveCallback(callback));
     }
@@ -193,9 +196,12 @@ namespace eCAL
     **/
     bool RemReceiveCallback()
     {
+      bool ret = CSubscriber::RemReceiveCallback();
+
+      std::lock_guard<std::mutex> callback_lock(m_cb_callback_mutex);
       if(m_cb_callback == nullptr) return(false);
       m_cb_callback = nullptr;
-      return(CSubscriber::RemReceiveCallback());
+      return(ret);
     }
 
   private:
@@ -205,7 +211,11 @@ namespace eCAL
 
     void ReceiveCallback(const char* topic_name_, const struct eCAL::SReceiveCallbackData* data_)
     {
-      MsgReceiveCallbackT fn_callback(m_cb_callback);
+      MsgReceiveCallbackT fn_callback = nullptr;
+      {
+        std::lock_guard<std::mutex> callback_lock(m_cb_callback_mutex);
+        fn_callback = m_cb_callback;
+      }
 
       if(fn_callback == nullptr) return;
 
@@ -216,6 +226,7 @@ namespace eCAL
       }
     }
 
+    std::mutex          m_cb_callback_mutex;
     MsgReceiveCallbackT m_cb_callback;
   };
 }
diff --git a/testing/ecal/core_test/src/core_test.cpp b/testing/ecal/core_test/src/core_test.cpp
@@ -5,9 +5,9 @@
  * 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.
@@ -96,7 +96,7 @@ TEST(Core, LeakedPubSub)
       std::this_thread::sleep_for(std::chrono::milliseconds(100));
 #endif
     }
-  });
+    });
 
   // let them work together
   std::this_thread::sleep_for(std::chrono::seconds(2));
@@ -111,66 +111,67 @@ TEST(Core, LeakedPubSub)
 
 TEST(Core, CallbackDestruction)
 {
-  // initialize eCAL API
-  EXPECT_EQ(0, eCAL::Initialize(0, nullptr, "callback destruction"));
+  for (int i = 0; i < 10; ++i)
+  {
+    // initialize eCAL API
+    EXPECT_EQ(0, eCAL::Initialize(0, nullptr, "callback destruction"));
 
-  // enable loop back communication in the same thread
-  eCAL::Util::EnableLoopback(true);
+    // enable loop back communication in the same thread
+    eCAL::Util::EnableLoopback(true);
 
-  // create subscriber and register a callback
-  std::shared_ptr< eCAL::string::CSubscriber<std::string>> sub;
+    // create subscriber and register a callback
+    std::shared_ptr<eCAL::string::CSubscriber<std::string>> sub;
 
-  // create publisher
-  eCAL::string::CPublisher<std::string> pub("foo");
+    // create publisher
+    eCAL::string::CPublisher<std::string> pub("foo");
 
-  // start publishing thread
-  std::atomic<bool> pub_stop(false);
-  std::thread pub_t([&]() {
-    while (!pub_stop)
-    {
-      pub.Send("Hello World");
+    // start publishing thread
+    std::atomic<bool> pub_stop(false);
+    std::thread pub_t([&]() {
+      while (!pub_stop) {
+        pub.Send("Hello World");
 #if 0
-      // some kind of busy waiting....
-      int y = 0;
-      for (int i = 0; i < 100000; i++)
-      {
-        y += i;
-      }
+        // some kind of busy waiting....
+        int y = 0;
+        for (int i = 0; i < 100000; i++)
+        {
+          y += i;
+        }
 #else
-      std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
 #endif
-    }
-    });
-
-  std::atomic<bool> sub_stop(false);
-  std::thread sub_t([&]() {
-    while (!sub_stop)
-    {
-      sub = std::make_shared<eCAL::string::CSubscriber<std::string>>("foo");
-      sub->AddReceiveCallback(std::bind(OnReceive, std::placeholders::_4));
-      std::this_thread::sleep_for(std::chrono::seconds(2));
-    }
-    });
+      }
+      });
+
+    std::atomic<bool> sub_stop(false);
+    std::thread sub_t([&]() {
+      while (!sub_stop) {
+        sub = std::make_shared<eCAL::string::CSubscriber<std::string>>("foo");
+        sub->AddReceiveCallback(std::bind(OnReceive, std::placeholders::_4));
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+      }
+      });
 
-  // let them work together
-  std::this_thread::sleep_for(std::chrono::seconds(10));
+    // let them work together
+    std::this_thread::sleep_for(std::chrono::seconds(10));
 
-  // stop publishing thread
-  pub_stop = true;
-  pub_t.join();
+    // stop publishing thread
+    pub_stop = true;
+    pub_t.join();
 
-  sub_stop = true;
-  sub_t.join();
+    sub_stop = true;
+    sub_t.join();
 
-  // finalize eCAL API
-  // without destroying any pub / sub
-  EXPECT_EQ(0, eCAL::Finalize());
+    // finalize eCAL API
+    // without destroying any pub / sub
+    EXPECT_EQ(0, eCAL::Finalize());
+  }
 }
 
 /* excluded for now, system timer jitter too high */
 #if 0
 TEST(Core, TimerCallback)
-{ 
+{
   // initialize eCAL API
   EXPECT_EQ(0, eCAL::Initialize(0, nullptr, "timer callback"));