Merge pull request #77 from c-jimenez/fix/multithreading_issues

Fix/multithreading issues

Author: c-jimenez

.github/workflows/codeql-analysis.yml

@@ -42,7 +42,7 @@ jobs:
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v1
+      uses: github/codeql-action/init@v2
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -66,4 +66,4 @@ jobs:
        make clang-native
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v1
+      uses: github/codeql-action/analyze@v2

src/chargepoint/reservation/ReservationManager.cpp

@@ -311,7 +311,7 @@ void ReservationManager::checkExpiries()
         if ((connector->status == ChargePointStatus::Reserved) && (connector->reservation_expiry_date <= now))
         {
             // End reservation
-            endReservation(connector->id, false);
+            m_worker_pool.run<void>(std::bind(&ReservationManager::endReservation, this, connector->id, false));
         }
     }
 }

src/chargepoint/status/StatusManager.cpp

@@ -67,8 +67,8 @@ StatusManager::StatusManager(const ocpp::config::IChargePointConfig&         sta
       m_boot_notification_timer(timer_pool, "Boot notification"),
       m_heartbeat_timer(timer_pool, "Heartbeat")
 {
-    m_boot_notification_timer.setCallback(std::bind(&StatusManager::bootNotificationProcess, this));
-    m_heartbeat_timer.setCallback(std::bind(&StatusManager::heartBeatProcess, this));
+    m_boot_notification_timer.setCallback([this] { m_worker_pool.run<void>(std::bind(&StatusManager::bootNotificationProcess, this)); });
+    m_heartbeat_timer.setCallback([this] { m_worker_pool.run<void>(std::bind(&StatusManager::heartBeatProcess, this)); });
 
     trigger_manager.registerHandler(ocpp::types::MessageTrigger::BootNotification, *this);
     trigger_manager.registerHandler(ocpp::types::MessageTrigger::Heartbeat, *this);
@@ -178,7 +178,9 @@ bool StatusManager::updateConnectorStatus(unsigned int                      conn
                     connector->status_timer.stop();
                     if (connector->status != connector->last_notified_status)
                     {
-                        connector->status_timer.setCallback([connector_id, this] { statusNotificationProcess(connector_id); });
+                        connector->status_timer.setCallback(
+                            [this, connector_id]
+                            { m_worker_pool.run<void>(std::bind(&StatusManager::statusNotificationProcess, this, connector_id)); });
                         connector->status_timer.start(std::chrono::milliseconds(duration), true);
                     }
                 }

src/rpc/RpcClient.cpp

@@ -30,7 +30,7 @@ namespace rpc
 
 /** @brief Constructor */
 RpcClient::RpcClient(ocpp::websockets::IWebsocketClient& websocket, const std::string& protocol)
-    : RpcBase(), m_protocol(protocol), m_websocket(websocket), m_listener(nullptr), m_started(false)
+    : RpcBase(), m_protocol(protocol), m_websocket(websocket), m_listener(nullptr), m_started(false), m_stop_mutex()
 {
     m_websocket.registerListener(*this);
 }
@@ -71,15 +71,22 @@ bool RpcClient::stop()
 {
     bool ret = false;
 
-    // Check if already started
-    if (m_started)
+    // Check if someone is already stopping the client
+    // May happen in local controller mode when disconnection
+    // can be triggered from central system side and charge point side
+    if (m_stop_mutex.try_lock())
     {
-        // Disconnect from websocket
-        ret = m_websocket.disconnect();
+        // Check if already started
+        if (m_started)
+        {
+            // Disconnect from websocket
+            ret = m_websocket.disconnect();
 
-        // Stop processing
-        RpcBase::stop();
-        m_started = false;
+            // Stop processing
+            RpcBase::stop();
+            m_started = false;
+        }
+        m_stop_mutex.unlock();
     }
 
     return ret;

src/rpc/RpcClient.h

@@ -115,6 +115,8 @@ class RpcClient : public RpcBase, public ocpp::websockets::IWebsocketClient::ILi
     IListener* m_listener;
     /** @brief Started state */
     bool m_started;
+    /** @brief Mutex for concurrent stop access */
+    std::mutex m_stop_mutex;
 };
 
 } // namespace rpc

src/websockets/libwebsockets/LibWebsocketClient.cpp

@@ -18,6 +18,7 @@ along with OpenOCPP. If not, see <http://www.gnu.org/licenses/>.
 
 #include "LibWebsocketClient.h"
 
+#include <csignal>
 #include <cstdint>
 #include <functional>
 #include <iostream>
@@ -234,6 +235,12 @@ void LibWebsocketClient::process()
     // Save this pointer for further callbacks
     client = this;
 
+    // Mask SIG_PIPE signal
+    sigset_t set;
+    sigemptyset(&set);
+    sigaddset(&set, SIGPIPE);
+    pthread_sigmask(SIG_BLOCK, &set, NULL);
+
     // Event loop
     int ret = 0;
     while (!m_end && (ret >= 0))

src/websockets/libwebsockets/LibWebsocketServer.cpp

@@ -18,6 +18,7 @@ along with OpenOCPP. If not, see <http://www.gnu.org/licenses/>.
 
 #include "LibWebsocketServer.h"
 
+#include <csignal>
 #include <cstdint>
 #include <functional>
 #include <iostream>
@@ -222,6 +223,12 @@ void LibWebsocketServer::process()
     // Save this pointer for further callbacks
     server = this;
 
+    // Mask SIG_PIPE signal
+    sigset_t set;
+    sigemptyset(&set);
+    sigaddset(&set, SIGPIPE);
+    pthread_sigmask(SIG_BLOCK, &set, NULL);
+
     // Event loop
     int ret = 0;
     while (!m_end && (ret >= 0))