Merge pull request #45098 from smorovic/140x-fix-contention

fix lock contention in source (14_0_X backport)
cms-sw · Jun 3, 2024 · 6f041f7 · 6f041f7
2 parents facff27 + 5c7f19b
commit 6f041f7
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 11 deletions.
diff --git a/EventFilter/Utilities/src/DAQSource.cc b/EventFilter/Utilities/src/DAQSource.cc
@@ -1075,12 +1075,13 @@ void DAQSource::readWorker(unsigned int tid) {
     workerPool_.push(tid);
 
     if (init) {
-      std::unique_lock<std::mutex> lk(startupLock_);
+      std::unique_lock<std::mutex> lks(startupLock_);
       init = false;
       startupCv_.notify_one();
     }
     cvWakeup_.notify_all();
     cvReader_[tid]->wait(lk);
+    lk.unlock();
 
     if (thread_quit_signal[tid])
       return;

diff --git a/EventFilter/Utilities/src/EvFDaqDirector.cc b/EventFilter/Utilities/src/EvFDaqDirector.cc
@@ -24,6 +24,7 @@
 #include <unistd.h>
 #include <cstdio>
 #include <boost/algorithm/string.hpp>
+#include <fmt/printf.h>
 
 //using boost::asio::ip::tcp;
 
@@ -1549,6 +1550,7 @@ namespace evf {
                                                                int maxLS) {
     EvFDaqDirector::FileStatus fileStatus = noFile;
     serverError = false;
+    std::string dest = fmt::sprintf(" on connection to %s:%s", fileBrokerHost_, fileBrokerPort_);
 
     boost::system::error_code ec;
     try {
@@ -1558,7 +1560,7 @@ namespace evf {
           boost::asio::connect(*socket_, *endpoint_iterator_, ec);
 
           if (ec) {
-            edm::LogWarning("EvFDaqDirector") << "boost::asio::connect error -:" << ec;
+            edm::LogWarning("EvFDaqDirector") << "boost::asio::connect error -:" << ec << dest;
             serverError = true;
             break;
           }
@@ -1581,25 +1583,25 @@ namespace evf {
         boost::asio::write(*socket_, request, ec);
         if (ec) {
           if (fileBrokerKeepAlive_ && ec == boost::asio::error::connection_reset) {
-            edm::LogInfo("EvFDaqDirector") << "reconnecting socket on received connection_reset";
+            edm::LogInfo("EvFDaqDirector") << "reconnecting socket on received connection_reset" << dest;
             //we got disconnected, try to reconnect to the server before writing the request
             boost::asio::connect(*socket_, *endpoint_iterator_, ec);
             if (ec) {
-              edm::LogWarning("EvFDaqDirector") << "boost::asio::connect error -:" << ec;
+              edm::LogWarning("EvFDaqDirector") << "boost::asio::connect error -:" << ec << dest;
               serverError = true;
               break;
             }
             continue;
           }
-          edm::LogWarning("EvFDaqDirector") << "boost::asio::write error -:" << ec;
+          edm::LogWarning("EvFDaqDirector") << "boost::asio::write error -:" << ec << dest;
           serverError = true;
           break;
         }
 
         boost::asio::streambuf response;
         boost::asio::read_until(*socket_, response, "\r\n", ec);
         if (ec) {
-          edm::LogWarning("EvFDaqDirector") << "boost::asio::read_until error -:" << ec;
+          edm::LogWarning("EvFDaqDirector") << "boost::asio::read_until error -:" << ec << dest;
           serverError = true;
           break;
         }
@@ -1768,7 +1770,7 @@ namespace evf {
           while (boost::asio::read(*socket_, response, boost::asio::transfer_at_least(1), ec)) {
           }
           if (ec != boost::asio::error::eof) {
-            edm::LogWarning("EvFDaqDirector") << "boost::asio::read_until error -:" << ec;
+            edm::LogWarning("EvFDaqDirector") << "boost::asio::read_until error -:" << ec << dest;
             serverError = true;
           }
         }
@@ -1784,19 +1786,19 @@ namespace evf {
     if (!fileBrokerKeepAlive_ && socket_->is_open()) {
       socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
       if (ec) {
-        edm::LogWarning("EvFDaqDirector") << "socket shutdown error -:" << ec;
+        edm::LogWarning("EvFDaqDirector") << "socket shutdown error -:" << ec << dest;
       }
       socket_->close(ec);
       if (ec) {
-        edm::LogWarning("EvFDaqDirector") << "socket close error -:" << ec;
+        edm::LogWarning("EvFDaqDirector") << "socket close error -:" << ec << dest;
       }
     }
 
     if (serverError) {
       if (socket_->is_open())
         socket_->close(ec);
       if (ec) {
-        edm::LogWarning("EvFDaqDirector") << "socket close error -:" << ec;
+        edm::LogWarning("EvFDaqDirector") << "socket close error -:" << ec << dest;
       }
       fileStatus = noFile;
       sleep(1);  //back-off if error detected

diff --git a/EventFilter/Utilities/src/FedRawDataInputSource.cc b/EventFilter/Utilities/src/FedRawDataInputSource.cc
@@ -1299,12 +1299,13 @@ void FedRawDataInputSource::readWorker(unsigned int tid) {
     workerPool_.push(tid);
 
     if (init) {
-      std::unique_lock<std::mutex> lk(startupLock_);
+      std::unique_lock<std::mutex> lks(startupLock_);
       init = false;
       startupCv_.notify_one();
     }
     cvWakeup_.notify_all();
     cvReader_[tid]->wait(lk);
+    lk.unlock();
 
     if (thread_quit_signal[tid])
       return;