Fix data race in AbstractInvoker

src/framework/global/thirdparty/kors_async/async/internal/abstractinvoker.cpp

@@ -24,6 +24,7 @@ SOFTWARE.
 #include "abstractinvoker.h"
 
 #include <cassert>
+#include <iostream>
 
 #include "queuedinvoker.h"
 
@@ -39,6 +40,8 @@ AbstractInvoker::~AbstractInvoker()
     for (QInvoker* qi : m_qInvokers) {
         qi->invalidate();
     }
+
+    m_qInvokers.clear();
 }
 
 void AbstractInvoker::invoke(int type)
@@ -48,21 +51,37 @@ void AbstractInvoker::invoke(int type)
 
 void AbstractInvoker::invoke(int type, const NotifyData& data)
 {
-    auto it = m_callbacks.find(type);
-    if (it == m_callbacks.end()) {
-        return;
+    CallBacks callbacks;
+    {
+        std::shared_lock lock(m_callbacksMutex);
+
+        auto it = m_callbacks.find(type);
+        if (it == m_callbacks.end()) {
+            return;
+        }
+
+        //! NOTE: explicit copy because collection can be modified from elsewhere
+        callbacks = it->second;
     }
 
     std::thread::id threadID = std::this_thread::get_id();
 
-    //! NOTE: explicit copy because collection can be modified from elsewhere
-    CallBacks callbacks = it->second;
-
     for (const CallBack& c : callbacks) {
-        if (!it->second.containsReceiver(c.receiver)) {
-            std::cout << "Skipping removed receiver";
-            continue;
+        {
+            std::shared_lock lock(m_callbacksMutex);
+
+            auto it = m_callbacks.find(type);
+            if (it == m_callbacks.end()) {
+                std::cout << "Skipping removed callback type";
+                continue;
+            }
+
+            if (!it->second.containsReceiver(c.receiver)) {
+                std::cout << "Skipping removed receiver";
+                continue;
+            }
         }
+
         if (c.threadID == threadID) {
             invokeCallback(type, c, data);
         } else {
@@ -102,6 +121,8 @@ void AbstractInvoker::onMainThreadInvoke(const std::function<void(const std::fun
 
 bool AbstractInvoker::isConnected() const
 {
+    std::shared_lock lock(m_callbacksMutex);
+
     for (auto it = m_callbacks.cbegin(); it != m_callbacks.cend(); ++it) {
         const CallBacks& cs = it->second;
         if (cs.size() > 0) {
@@ -127,6 +148,12 @@ bool AbstractInvoker::CallBacks::containsReceiver(Asyncable* receiver) const
 }
 
 void AbstractInvoker::removeCallBack(int type, Asyncable* receiver)
+{
+    std::unique_lock lock(m_callbacksMutex);
+    doRemoveCallBack(type, receiver);
+}
+
+void AbstractInvoker::doRemoveCallBack(int type, Asyncable* receiver)
 {
     auto it = m_callbacks.find(type);
     if (it == m_callbacks.end()) {
@@ -147,9 +174,11 @@ void AbstractInvoker::removeCallBack(int type, Asyncable* receiver)
 
     {
         std::lock_guard<std::mutex> lock(m_qInvokersMutex);
-        for (QInvoker* qi : m_qInvokers) {
+        for (auto it = m_qInvokers.begin(); it != m_qInvokers.end(); ++it) {
+            QInvoker* qi = *it;
             if (qi->call.call == c.call) {
                 qi->invalidate();
+                m_qInvokers.erase(it);
                 break;
             }
         }
@@ -160,6 +189,8 @@ void AbstractInvoker::removeCallBack(int type, Asyncable* receiver)
 
 void AbstractInvoker::removeAllCallBacks()
 {
+    std::unique_lock lock(m_callbacksMutex);
+
     for (auto it = m_callbacks.begin(); it != m_callbacks.end(); ++it) {
         for (CallBack& c : it->second) {
             if (c.receiver) {
@@ -174,14 +205,16 @@ void AbstractInvoker::removeAllCallBacks()
 
 void AbstractInvoker::addCallBack(int type, Asyncable* receiver, void* call, Asyncable::AsyncMode mode)
 {
+    std::unique_lock lock(m_callbacksMutex);
+
     const CallBacks& callbacks = m_callbacks[type];
     if (callbacks.containsReceiver(receiver)) {
         switch (mode) {
         case Asyncable::AsyncMode::AsyncSetOnce:
             deleteCall(type, call);
             return;
         case Asyncable::AsyncMode::AsyncSetRepeat:
-            removeCallBack(type, receiver);
+            doRemoveCallBack(type, receiver);
             break;
         }
     }
@@ -196,6 +229,8 @@ void AbstractInvoker::addCallBack(int type, Asyncable* receiver, void* call, Asy
 
 void AbstractInvoker::disconnectAsync(Asyncable* receiver)
 {
+    std::unique_lock lock(m_callbacksMutex);
+
     std::vector<int> types;
     for (auto it = m_callbacks.begin(); it != m_callbacks.end(); ++it) {
         for (CallBack& c : it->second) {
@@ -206,7 +241,7 @@ void AbstractInvoker::disconnectAsync(Asyncable* receiver)
     }
 
     for (int type : types) {
-        removeCallBack(type, receiver);
+        doRemoveCallBack(type, receiver);
     }
 }
 
@@ -224,6 +259,8 @@ void AbstractInvoker::removeQInvoker(QInvoker* qi)
 
 bool AbstractInvoker::containsReceiver(Asyncable* receiver) const
 {
+    std::shared_lock lock(m_callbacksMutex);
+
     for (auto it = m_callbacks.begin(); it != m_callbacks.end(); ++it) {
         for (const CallBack& c : it->second) {
             if (c.receiver == receiver) {

src/framework/global/thirdparty/kors_async/async/internal/abstractinvoker.h

@@ -27,9 +27,9 @@ SOFTWARE.
 #include <memory>
 #include <vector>
 #include <list>
-#include <iostream>
 #include <map>
 #include <mutex>
+#include <shared_mutex>
 #include <thread>
 #include <functional>
 
@@ -106,6 +106,11 @@ class AbstractInvoker : public Asyncable::IConnectable
     virtual void deleteCall(int type, void* call) = 0;
     virtual void doInvoke(int type, void* call, const NotifyData& data) = 0;
 
+    void addCallBack(int type, Asyncable* receiver, void* call, Asyncable::AsyncMode mode = Asyncable::AsyncMode::AsyncSetRepeat);
+    void removeCallBack(int type, Asyncable* receiver);
+    void removeAllCallBacks();
+
+private:
     struct CallBack {
         std::thread::id threadID;
         int type = 0;
@@ -166,15 +171,14 @@ class AbstractInvoker : public Asyncable::IConnectable
 
     void invokeCallback(int type, const CallBack& c, const NotifyData& data);
 
-    void addCallBack(int type, Asyncable* receiver, void* call, Asyncable::AsyncMode mode = Asyncable::AsyncMode::AsyncSetRepeat);
-    void removeCallBack(int type, Asyncable* receiver);
-    void removeAllCallBacks();
+    void doRemoveCallBack(int type, Asyncable* receiver);
 
     void addQInvoker(QInvoker* qi);
     void removeQInvoker(QInvoker* qi);
 
     bool containsReceiver(Asyncable* receiver) const;
 
+    mutable std::shared_mutex m_callbacksMutex;
     std::map<int /*type*/, CallBacks > m_callbacks;
 
     std::mutex m_qInvokersMutex;

src/framework/global/thirdparty/kors_modularity/modularity/ioc.cpp

@@ -23,7 +23,7 @@ SOFTWARE.
 */
 #include "ioc.h"
 
-std::mutex kors::modularity::StaticMutex::mutex;
+std::shared_mutex kors::modularity::StaticMutex::mutex;
 
 static std::map<kors::modularity::IoCID, kors::modularity::ModulesIoC*> s_map;
 

src/framework/global/thirdparty/kors_modularity/modularity/ioc.h

@@ -25,7 +25,7 @@ SOFTWARE.
 #define KORS_MODULARITY_IOC_H
 
 #include <memory>
-#include <mutex>
+#include <shared_mutex>
 
 #include "context.h"
 #include "modulesioc.h"
@@ -37,7 +37,7 @@ void removeIoC(const ContextPtr& ctx = nullptr);
 
 struct StaticMutex
 {
-    static std::mutex mutex;
+    static std::shared_mutex mutex;
 };
 
 template<class I>
@@ -73,18 +73,28 @@ class Inject
 
     const std::shared_ptr<I>& get() const
     {
-        if (!m_i) {
-            const std::lock_guard<std::mutex> lock(StaticMutex::mutex);
-            if (!m_i) {
-                static std::string_view module = "";
-                m_i = _ioc(iocContext())->template resolve<I>(module);
+        {
+            std::shared_lock lock(StaticMutex::mutex);
+            if (m_i) {
+                return m_i;
             }
         }
+
+        std::unique_lock lock(StaticMutex::mutex);
+
+        // Need to check again, because between the release of the shared lock and the acquisition
+        // of the unique lock, another thread could have already created the instance.
+        if (!m_i) {
+            static std::string_view module = "";
+            m_i = _ioc(iocContext())->template resolve<I>(module);
+        }
+
         return m_i;
     }
 
     void set(std::shared_ptr<I> impl)
     {
+        std::unique_lock lock(StaticMutex::mutex);
         m_i = impl;
     }