[Core] Fix Race Conditions in CMsgSubscriber

ecal/core/include/ecal/msg/capnproto/subscriber.h

@@ -68,6 +68,14 @@ namespace eCAL
       {
       }
 
+      /**
+       * @brief  Destructor
+      **/
+      ~CBuilderSubscriber() override
+      {
+        this->Destroy();
+      }
+
       /**
       * @brief  Copy Constructor is not available.
       **/

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.
       **/

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.
       **/

ecal/core/include/ecal/msg/protobuf/subscriber.h

@@ -74,6 +74,14 @@ namespace eCAL
       {
       }
 
+      /**
+       * @brief  Destructor
+      **/
+      ~CSubscriber() override
+      {
+        this->Destroy();
+      }
+
       /**
       * @brief  Copy Constructor is not available.
       **/

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.
       **/

ecal/core/include/ecal/msg/subscriber.h

@@ -76,6 +76,8 @@ namespace eCAL
     {
     }
 
+    virtual ~CMsgSubscriber() = default;
+
     /**
     * @brief  Copy Constructor is not available.
     **/
@@ -125,8 +127,6 @@ namespace eCAL
       return *this;
     }
 
-    virtual ~CMsgSubscriber() {}
-
     /**
      * @brief Creates this object.
      *
@@ -207,7 +207,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));
     }
@@ -219,9 +222,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);
     }
 
 protected:
@@ -245,7 +251,11 @@ namespace eCAL
   private:
     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;
 
@@ -256,6 +266,7 @@ namespace eCAL
       }
     }
 
+    std::mutex          m_cb_callback_mutex;
     MsgReceiveCallbackT m_cb_callback;
   };
 }

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"));