Tsan failure fixes (#964)

* pull in some changes from CIRCLE_CI_FIXES branch

* update containers library

* add in locking for transmitters

* fix some other errors with `this->` in  lambda

* remove use of nullptr for insert in ValueFederateManager

* fix incorrect unlock call in connect method for commsInterface

* add a thread join in the destructor oc commsInterface to make sure the join is called.

* add a shared_from_this into an additional callback for tcpconnections

* update concurrency library with a different fix for triggerVariable

* switch tsan to the clang-8 version

* move octave build/test to buildenv image

* Automated formatting of source files (#965)

* bump concurrency to a tagged version

* Tweak the formatting of one of the lambdas

.circleci/config.yml

@@ -47,7 +47,7 @@ aliases:
 jobs:
   helicsoctave:
     docker:
-      - image: helics/octave:latest
+      - image: helics/buildenv:octave
     environment:
       OCTAVETEST: "/root/project/tests/octave"
       CMAKE_FLAGS: "-DBUILD_OCTAVE_INTERFACE=ON -DHELICS_BUILD_TESTS=ON"
@@ -61,7 +61,7 @@ jobs:
       - image: helics/clang-tsan:latest
     environment:
       CMAKE_FLAGS: '-DCMAKE_CXX_COMPILER=clang++ -DCMAKE_C_COMPILER=clang -DCMAKE_CXX_FLAGS="-fsanitize=thread -stdlib=libc++ -L/root/develop/libcxx_tsan/lib -lc++abi -I/root/develop/libcxx_tsan/include -I/root/develop/libcxx_tsan/include/c++/v1 -Wno-unused-command-line-argument -g -O1 -Wl,-rpath=/root/develop/libcxx_tsan/lib"  -DHELICS_BUILD_TESTS=ON -DHELICS_ZMQ_SUBPROJECT=ON  -DHELICS_ZMQ_FORCE_SUBPROJECT=ON -DHELICS_DISABLE_SYSTEM_CALL_TESTS=ON'
-      TSAN_OPTIONS: 'second_deadlock_stack=1 suppressions=/root/project/.circleci/tsan_suppression.txt'
+      TSAN_OPTIONS: 'second_deadlock_stack=1 suppressions=/root/project/.circleci/tsan_suppression.txt history_size=4'
 
     steps:
       - checkout

.clang-format

@@ -19,7 +19,7 @@ AllowShortLoopsOnASingleLine: 'false'
 AlwaysBreakAfterDefinitionReturnType: None
 AlwaysBreakAfterReturnType: None
 AlwaysBreakBeforeMultilineStrings: 'false'
-AlwaysBreakTemplateDeclarations: 'Yes'
+AlwaysBreakTemplateDeclarations: 'true'
 BinPackArguments: 'false'
 BinPackParameters: 'false'
 BreakBeforeBraces: Custom

src/helics/application_api/ValueFederateManager.cpp

@@ -61,7 +61,7 @@ Publication& ValueFederateManager::registerPublication(
     if (!key.empty()) {
         active = pubHandle->insert(key, coreID, fed, coreID, key, type, units);
     } else {
-        active = pubHandle->insert(nullptr, coreID, fed, coreID, key, type, units);
+        active = pubHandle->insert(no_search, coreID, fed, coreID, key, type, units);
     }
 
     if (active) {
@@ -81,7 +81,7 @@ Input& ValueFederateManager::registerInput(
     if (!key.empty()) {
         active = inpHandle->insert(key, coreID, fed, coreID, key, units);
     } else {
-        active = inpHandle->insert(nullptr, coreID, fed, coreID, key, units);
+        active = inpHandle->insert(no_search, coreID, fed, coreID, key, units);
     }
     if (active) {
         auto& ref = inpHandle->back();
@@ -487,9 +487,7 @@ int ValueFederateManager::getInputCount() const
 
 void ValueFederateManager::clearUpdates()
 {
-    for (auto& inp : inputs.lock()) {
-        inp.clearUpdate();
-    }
+    inputs.lock()->apply([](auto& inp) { inp.clearUpdate(); });
 }
 
 void ValueFederateManager::clearUpdate(const Input& inp)

src/helics/core/CommonCore.cpp

@@ -2119,7 +2119,7 @@ void CommonCore::processPriorityCommand(ActionMessage&& command)
         case CMD_REG_FED: {
             // this one in the core needs to be the thread-safe version
             auto fed = getFederate(command.name);
-            loopFederates.insert(command.name, nullptr, fed);
+            loopFederates.insert(command.name, no_search, fed);
         }
             if (global_broker_id_local != parent_broker_id) {
                 // forward on to Broker
@@ -2296,11 +2296,11 @@ void CommonCore::sendErrorToFederates(int error_code)
 {
     ActionMessage errorCom(CMD_ERROR);
     errorCom.messageID = error_code;
-    for (auto& fed : loopFederates) {
+    loopFederates.apply([&errorCom](auto& fed) {
         if ((fed) && (!fed.disconnected)) {
             fed->addAction(errorCom);
         }
-    }
+    });
 }
 
 void CommonCore::transmitDelayedMessages(global_federate_id source)
@@ -2411,9 +2411,7 @@ void CommonCore::processCommand(ActionMessage&& command)
             break;
         case CMD_BROADCAST_DISCONNECT: {
             timeCoord->processTimeMessage(command);
-            for (auto& fed : loopFederates) {
-                fed->addAction(command);
-            }
+            loopFederates.apply([&command](auto& fed) { fed->addAction(command); });
             checkAndProcessDisconnect();
         } break;
         case CMD_STOP:
@@ -2686,9 +2684,7 @@ void CommonCore::processCommand(ActionMessage&& command)
             if (brokerState.compare_exchange_strong(
                     exp, broker_state_t::operating)) { // forward the grant to all federates
                 organizeFilterOperations();
-                for (auto& fed : loopFederates) {
-                    fed->addAction(command);
-                }
+                loopFederates.apply([&command](auto& fed) { fed->addAction(command); });
                 timeCoord->enteringExecMode();
                 auto res = timeCoord->checkExecEntry();
                 if (res == message_processing_result::next_step) {
@@ -3176,7 +3172,7 @@ void CommonCore::checkDependencies()
 {
     bool isobs = false;
     bool issource = false;
-    for (auto& fed : loopFederates) {
+    auto checkdep = [this, &isobs, &issource](auto& fed) {
         if (fed->endpointCount() > 0) {
             if (fed->getOptionFlag(defs::flags::observer)) {
                 timeCoord->removeDependency(fed->global_id);
@@ -3196,7 +3192,8 @@ void CommonCore::checkDependencies()
                 issource = true;
             }
         }
-    }
+    };
+    loopFederates.apply(checkdep);
 
     // if the core has filters we need to be a timeCoordinator
     if (hasFilters) {
@@ -3412,14 +3409,14 @@ void CommonCore::processCommandsForCore(const ActionMessage& cmd)
                 }
                 ActionMessage bye(CMD_DISCONNECT_FED_ACK);
                 bye.source_id = parent_broker_id;
-                for (auto& fed : loopFederates) {
+                loopFederates.apply([&bye](auto& fed) {
                     auto state = fed->getState();
                     if ((HELICS_FINISHED == state) || (HELICS_ERROR == state)) {
-                        continue;
+                        return;
                     }
                     bye.dest_id = fed->global_id.load();
                     fed->addAction(bye);
-                }
+                });
 
                 addActionMessage(CMD_STOP);
             } else {

src/helics/core/CommsInterface.cpp

@@ -19,15 +19,7 @@ CommsInterface::CommsInterface(thread_generation threads):
 /** destructor*/
 CommsInterface::~CommsInterface()
 {
-    std::lock_guard<std::mutex> syncLock(threadSyncLock);
-    if (!singleThread) {
-        if (queue_watcher.joinable()) {
-            queue_watcher.join();
-        }
-    }
-    if (queue_transmitter.joinable()) {
-        queue_transmitter.join();
-    }
+    join_tx_rx_thread();
 }
 
 void CommsInterface::loadNetworkInfo(const NetworkBrokerData& netInfo)
@@ -227,18 +219,28 @@ bool CommsInterface::connect()
             logError(std::string("error in transmitter >") + e.what());
         }
     });
+    syncLock.unlock();
     txTrigger.waitActivation();
     rxTrigger.waitActivation();
     if (rx_status != connection_status::connected) {
         logError("receiver connection failure");
         if (tx_status == connection_status::connected) {
+            syncLock.lock();
             if (queue_transmitter.joinable()) {
+                syncLock.unlock();
                 closeTransmitter();
-                queue_transmitter.join();
+                syncLock.lock();
+                if (queue_transmitter.joinable()) {
+                    queue_transmitter.join();
+                }
             }
+            syncLock.unlock();
         }
         if (!singleThread) {
-            queue_watcher.join();
+            syncLock.lock();
+            if (queue_watcher.joinable()) {
+                queue_watcher.join();
+            }
         }
         return false;
     }
@@ -247,13 +249,22 @@ bool CommsInterface::connect()
         logError("transmitter connection failure");
         if (!singleThread) {
             if (rx_status == connection_status::connected) {
+                syncLock.lock();
                 if (queue_watcher.joinable()) {
+                    syncLock.unlock();
                     closeReceiver();
-                    queue_watcher.join();
+                    syncLock.lock();
+                    if (queue_watcher.joinable()) {
+                        queue_watcher.join();
+                    }
                 }
+                syncLock.unlock();
             }
         }
-        queue_transmitter.join();
+        syncLock.lock();
+        if (queue_transmitter.joinable()) {
+            queue_transmitter.join();
+        }
         return false;
     }
     return true;
@@ -273,6 +284,7 @@ void CommsInterface::disconnect()
         if (propertyLock()) {
             setRxStatus(connection_status::terminated);
             setTxStatus(connection_status::terminated);
+            join_tx_rx_thread();
             return;
         }
     }
@@ -335,6 +347,20 @@ void CommsInterface::disconnect()
             return;
         }
     }
+    join_tx_rx_thread();
+}
+
+void CommsInterface::join_tx_rx_thread()
+{
+    std::lock_guard<std::mutex> syncLock(threadSyncLock);
+    if (!singleThread) {
+        if (queue_watcher.joinable()) {
+            queue_watcher.join();
+        }
+    }
+    if (queue_transmitter.joinable()) {
+        queue_transmitter.join();
+    }
 }
 
 bool CommsInterface::reconnect()

src/helics/core/CommsInterface.hpp

@@ -164,6 +164,8 @@ class CommsInterface {
     these functions should be called in a pair*/
     bool propertyLock();
     void propertyUnLock();
+    /** function to join the processing threads*/
+    void join_tx_rx_thread();
 
   private:
     gmlc::concurrency::TripWireDetector

src/helics/core/CoreBroker.cpp

@@ -262,7 +262,7 @@ void CoreBroker::processPriorityCommand(ActionMessage&& command)
                 transmit(getRoute(command.source_id), badName);
                 return;
             }
-            _federates.insert(command.name, nullptr, command.name);
+            _federates.insert(command.name, no_search, command.name);
             _federates.back().route = getRoute(command.source_id);
             _federates.back().parent = command.source_id;
             if (!isRootc) {
@@ -376,7 +376,7 @@ void CoreBroker::processPriorityCommand(ActionMessage&& command)
                 }
                 return;
             }
-            auto inserted = _brokers.insert(command.name, nullptr, command.name);
+            auto inserted = _brokers.insert(command.name, no_search, command.name);
             if (!inserted) {
                 route_id newroute;
                 bool route_created = false;
@@ -482,11 +482,12 @@ void CoreBroker::processPriorityCommand(ActionMessage&& command)
                 higher_broker_id = command.source_id;
                 timeCoord->source_id = global_broker_id_local;
                 transmitDelayedMessages();
-                for (auto& brk : _brokers) {
+                _brokers.apply([localid = global_broker_id_local](auto& brk) {
                     if (!brk._nonLocal) {
-                        brk.parent = global_broker_id_local;
+                        brk.parent = localid;
                     }
-                }
+                });
+
                 timeoutMon->setParentId(higher_broker_id);
                 timeoutMon->reset();
                 return;
@@ -620,34 +621,31 @@ void CoreBroker::transmitDelayedMessages()
 
 void CoreBroker::labelAsDisconnected(global_broker_id brkid)
 {
-    for (auto& brk : _brokers) {
-        if (brk.parent == brkid) {
-            brk.isDisconnected = true;
-        }
-    }
-    for (auto& fed : _federates) {
-        if (fed.parent == brkid) {
-            fed.isDisconnected = true;
+    auto disconnect_procedure = [brkid](auto& obj) {
+        if (obj.parent == brkid) {
+            obj.isDisconnected = true;
         }
-    }
+    };
+    _brokers.apply(disconnect_procedure);
+    _federates.apply(disconnect_procedure);
 }
 
 void CoreBroker::sendDisconnect()
 {
     ActionMessage bye(CMD_DISCONNECT);
     bye.source_id = global_broker_id_local;
-    for (auto& brk : _brokers) {
+    _brokers.apply([this, &bye](auto& brk) {
         if (!brk.isDisconnected) {
             if (brk.parent == global_broker_id_local) {
-                routeMessage(bye, brk.global_id);
+                this->routeMessage(bye, brk.global_id);
                 brk.isDisconnected = true;
             }
             if (hasTimeDependency) {
                 timeCoord->removeDependency(brk.global_id);
                 timeCoord->removeDependent(brk.global_id);
             }
         }
-    }
+    });
     if (hasTimeDependency) {
         timeCoord->disconnect();
     }
@@ -899,17 +897,17 @@ void CoreBroker::processCommand(ActionMessage&& command)
         case CMD_DISCONNECT_BROKER_ACK:
             if ((command.dest_id == global_broker_id_local) &&
                 (command.source_id == higher_broker_id)) {
-                for (auto& brk : _brokers) {
+                _brokers.apply([this](auto& brk) {
                     if (!brk._sent_disconnect_ack) {
                         ActionMessage dis(
                             (brk._core) ? CMD_DISCONNECT_CORE_ACK : CMD_DISCONNECT_BROKER_ACK);
                         dis.source_id = global_broker_id_local;
                         dis.dest_id = brk.global_id;
-                        transmit(brk.route, dis);
+                        this->transmit(brk.route, dis);
                         brk._sent_disconnect_ack = true;
-                        removeRoute(brk.route);
+                        this->removeRoute(brk.route);
                     }
-                }
+                });
                 addActionMessage(CMD_STOP);
             }
             break;

src/helics/core/tcp/TcpHelperClasses.cpp

@@ -37,8 +37,8 @@ namespace tcp {
             if (!triggerhalt) {
                 socket_.async_receive(
                     asio::buffer(data.data() + residBufferSize, data.size() - residBufferSize),
-                    [this](const std::error_code& error, size_t bytes_transferred) {
-                        handle_read(error, bytes_transferred);
+                    [ptr = shared_from_this()](const std::error_code& err, size_t bytes) {
+                        ptr->handle_read(err, bytes);
                     });
                 if (triggerhalt) {
                     // cancel previous operation if triggerhalt is now active

src/helics/shared_api_library/.clang-format

@@ -19,7 +19,7 @@ AllowShortLoopsOnASingleLine: 'false'
 AlwaysBreakAfterDefinitionReturnType: None
 AlwaysBreakAfterReturnType: None
 AlwaysBreakBeforeMultilineStrings: 'false'
-AlwaysBreakTemplateDeclarations: 'Yes'
+AlwaysBreakTemplateDeclarations: 'true'
 BinPackArguments: 'false'
 BinPackParameters: 'false'
 BreakBeforeBraces: Custom

tests/helics/system_tests/QueryTests.cpp

@@ -451,4 +451,4 @@ TEST_F(query_tests, test_queries_query)
     vFed1->finalize();
     vFed2->finalize();
     helics::cleanupHelicsLibrary();
-}
+}

tests/helics/system_tests/networkTests.cpp

@@ -118,4 +118,4 @@ TEST_F(network_tests, test_external_udp_ipv4)
     vFed1->finalize();
 }
 
-#endif
+#endif