option to protect shared memory with a named semaphore

Author: NikolasK-source

.gitmodules

@@ -7,3 +7,6 @@
 [submodule "libs/cxxshm"]
 	path = libs/cxxshm
 	url = https://github.com/NikolasK-source/cxxshm.git
+[submodule "libs/cxxsemaphore"]
+	path = libs/cxxsemaphore
+	url = https://github.com/NikolasK-source/cxxsemaphore.git

libs/CMakeLists.txt

@@ -13,3 +13,6 @@ target_link_libraries(${Target} PRIVATE cxxopts)
 
 add_subdirectory(cxxshm EXCLUDE_FROM_ALL)
 target_link_libraries(${Target} PRIVATE cxxshm)
+
+add_subdirectory(cxxsemaphore EXCLUDE_FROM_ALL)
+target_link_libraries(${Target} PRIVATE cxxsemaphore)

libs/cxxsemaphore

@@ -21,6 +21,9 @@ namespace TCP {
 
 static constexpr int MAX_REGS = 0x10000;
 
+//* maximum time to wait for semaphore (100ms)
+static constexpr struct timespec SEMAPHORE_MAX_TIME = {0, 100'000};
+
 Client_Poll::Client_Poll(const std::string &host,
                          const std::string &service,
                          modbus_mapping_t  *mapping,
@@ -174,6 +177,12 @@ void Client_Poll::set_tcp_timeout(std::size_t tcp_timeout) {
 }
 #endif
 
+void Client_Poll::enable_semaphore(const std::string &name, bool force) {
+    if (semaphore) throw std::logic_error("semaphore already enabled");
+
+    semaphore = std::make_unique<cxxsemaphore::Semaphore>(name, 1, force);
+}
+
 void Client_Poll::set_debug(bool debug) {
     if (modbus_set_debug(modbus, debug)) {
         const std::string error_msg = modbus_strerror(errno);
@@ -369,7 +378,14 @@ Client_Poll::run_t Client_Poll::run(int signal_fd, bool reconnect, int timeout)
                     auto mapping = mappings[CLIENT_ID];
 
                     // handle request
+                    if (semaphore) {
+                        if (!semaphore->wait(SEMAPHORE_MAX_TIME))
+                            std::cerr << Print_Time::iso << " WARNING: Failed to acquire semaphore '"
+                                      << semaphore->get_name() << "' within 100ms." << std::endl;
+                    }
                     int ret = modbus_reply(modbus, query, rc, mapping);
+                    if (semaphore && semaphore->is_acquired()) semaphore->post();
+
                     if (ret == -1) {
                         std::cerr << Print_Time::iso << " ERROR: modbus_reply failed: " << modbus_strerror(errno)
                                   << std::endl;

src/Modbus_TCP_Client_poll.cpp

@@ -5,6 +5,8 @@
 #pragma once
 
 #include <cstddef>
+#include <cxxsemaphore.hpp>
+#include <memory>
 #include <modbus/modbus.h>
 #include <string>
 #include <sys/poll.h>
@@ -39,6 +41,8 @@ class Client_Poll {
     int                                  server_socket = -1;  //!< socket of the modbus connection
     std::unordered_map<int, std::string> client_addrs;
 
+    std::unique_ptr<cxxsemaphore::Semaphore> semaphore;
+
 public:
     /*! \brief create modbus client (TCP server)
      *
@@ -68,11 +72,19 @@ class Client_Poll {
                 std::size_t        tcp_timeout = 5,
                 std::size_t        max_clients = 1);
 
-    /*! \brief destroy the modbus client
-     *
+    /**
+     * @brief destroy the modbus client
      */
     ~Client_Poll();
 
+    /**
+     * @brief use the semaphore mechanism
+     *
+     * @param name name of the shared
+     * @param force use the semaphore even if it already exists
+     */
+    void enable_semaphore(const std::string &name, bool force = false);
+
     /*! \brief enable/disable debugging output
      *
      * @param debug true: enable debug output

src/Modbus_TCP_Client_poll.hpp

@@ -171,6 +171,14 @@ int main(int argc, char **argv) {
                          ("separate-all",
                           "like --separate, but for all client ids (creates 1028 shared memory files! "
                           "check/set 'ulimit -n' before using this option.)")
+                         ("semaphore",
+                          "protect the shared memory with a named semaphore against simultaneous access",
+                          cxxopts::value<std::string>())
+                          ("semaphore-force",
+                          "Force the use of the semaphore even if it already exists. "
+                          "Do not use this option per default! "
+                          "It should only be used if the semaphore of an improperly terminated instance continues "
+                          "to exist as an orphan and is no longer used.")
                          ("h,help",
                           "print usage")
                          ("version",
@@ -383,6 +391,16 @@ int main(int argc, char **argv) {
         return EX_SOFTWARE;
     }
 
+    // add semaphore if required
+    try {
+        if (args.count("semaphore")) {
+            client->enable_semaphore(args["semaphore"].as<std::string>(), args.count("semaphore-force"));
+        }
+    } catch (const std::system_error &e) {
+        std::cerr << Print_Time::iso << " ERROR: " << e.what() << std::endl;
+        return EX_SOFTWARE;
+    }
+
     auto RECONNECT = args.count("reconnect") != 0;
 
     std::cerr << Print_Time::iso << " INFO: Listening on " << client->get_listen_addr() << " for connections."