fmq kvals: refactor dd kv subscribe methods

src/ReadoutEmulator/CruMemoryHandler.h

@@ -16,6 +16,8 @@
 
 #include "Headers/DataHeader.h"
 
+#include <fairmq/FairMQUnmanagedRegion.h>
+
 #include <stack>
 #include <map>
 #include <vector>
@@ -30,9 +32,7 @@
 
 class FairMQUnmanagedRegion;
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 using namespace o2::header;
@@ -145,7 +145,7 @@ class CruMemoryHandler
   /// output data queue
   ConcurrentFifo<ReadoutLinkO2Data> mO2LinkDataQueue;
 };
-}
+
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_CRU_MEMORY_HANDLER_H_ */

src/ReadoutEmulator/ReadoutDevice.h

@@ -15,14 +15,13 @@
 
 #include <ReadoutDataModel.h>
 #include <Utilities.h>
+#include <FmqUtilities.h>
 
 #include <memory>
 #include <deque>
 #include <condition_variable>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 class ReadoutDevice : public DataDistDevice
@@ -84,7 +83,7 @@ class ReadoutDevice : public DataDistDevice
 
   RunningSamples<uint64_t, 8192> mFreeSuperpagesSamples;
 };
-}
+
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_READOUT_EMULATOR_DEVICE_H_ */

src/StfBuilder/StfBuilderDevice.h

@@ -22,6 +22,7 @@
 #include <SubTimeFrameFileSource.h>
 #include <ConcurrentQueue.h>
 #include <Utilities.h>
+#include <FmqUtilities.h>
 
 #include <deque>
 #include <memory>

src/StfSender/StfSenderDevice.h

@@ -22,6 +22,7 @@
 
 #include <SubTimeFrameFileSink.h>
 #include <Utilities.h>
+#include <FmqUtilities.h>
 
 #include <thread>
 #include <vector>

src/TfBuilder/TfBuilderDevice.h

@@ -24,6 +24,7 @@
 #include <SubTimeFrameFileSource.h>
 #include <ConcurrentQueue.h>
 #include <Utilities.h>
+#include <FmqUtilities.h>
 
 #include <deque>
 #include <mutex>

src/TfBuilder/runTfBuilderDevice.cxx

@@ -79,9 +79,11 @@ int main(int argc, char* argv[])
 
         if (pKey == "partition_id" || pKey == "partition-id" || pKey == "environment-id" || pKey == "environment_id") {
 
-          if (r.fConfig.GetProperty<std::string>("discovery-partition") == "") {
-            r.fConfig.SetProperty<std::string>("discovery-partition", pValue);
-            IDDLOG("Config::Subscribe received key-value pair. {}=<{}>", pKey, pValue);
+          IDDLOG("Config::PartitionIdSubscribe received key-value pair. {}=<{}>", pKey, pValue);
+
+          if (r.fConfig.GetProperty<std::string>(Config::OptionKeyDiscoveryPartition) == "") {
+            r.fConfig.SetProperty<std::string>(Config::OptionKeyDiscoveryPartition, pValue);
+            IDDLOG("Config::PartitionIdSubscribe changed to: {}", pValue);
           }
         }
       });

src/TfScheduler/TfSchedulerDevice.h

@@ -19,6 +19,7 @@
 #include <ConfigConsul.h>
 
 #include <Utilities.h>
+#include <FmqUtilities.h>
 
 #include <thread>
 

src/TfScheduler/TfSchedulerInstance.h

@@ -20,14 +20,15 @@
 #include <ConfigConsul.h>
 
 #include <Utilities.h>
+#include <FmqUtilities.h>
+
+#include <fairmq/ProgOptionsFwd.h>
 
 #include <vector>
 #include <map>
 #include <thread>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 class ConsulConfig;
@@ -64,7 +65,7 @@ class TfSchedulerInstanceHandler
   TfSchedulerInstanceRpcImpl mRpcServer;
 
 };
-}
+
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_TF_SCHEDULER_INSTANCE_H_ */

src/common/CMakeLists.txt

@@ -31,6 +31,7 @@ target_include_directories(common
 target_link_libraries(common
   PUBLIC
     base
+    fmqtools
     Boost::iostreams
     FairMQ::FairMQ
     AliceO2::Headers

src/common/SubTimeFrameFileSink.cxx

@@ -13,8 +13,11 @@
 
 #include "SubTimeFrameFileSink.h"
 #include "FilePathUtils.h"
+#include "FmqUtilities.h"
 #include "DataDistLogger.h"
 
+#include <fairmq/ProgOptions.h>
+
 #include <boost/algorithm/string/replace.hpp>
 #include <boost/program_options/options_description.hpp>
 #include <boost/filesystem.hpp>

src/common/SubTimeFrameFileSink.h

@@ -21,6 +21,8 @@
 
 #include <Headers/DataHeader.h>
 
+#include <fairmq/ProgOptionsFwd.h>
+
 #include <boost/program_options/options_description.hpp>
 #include <boost/filesystem.hpp>
 #include <fstream>
@@ -33,6 +35,8 @@ namespace DataDistribution
 
 namespace bpo = boost::program_options;
 
+class DataDistDevice;
+
 ////////////////////////////////////////////////////////////////////////////////
 /// SubTimeFrameFileSink
 ////////////////////////////////////////////////////////////////////////////////

src/common/base/CMakeLists.txt

@@ -20,4 +20,5 @@ target_link_libraries(base
     spdlog::spdlog
     Boost::filesystem
     AliceO2::InfoLogger
+    FairLogger::FairLogger
 )

src/common/base/Utilities.h

@@ -14,8 +14,6 @@
 #ifndef ALICEO2_DATADIST_UTILITIES_H_
 #define ALICEO2_DATADIST_UTILITIES_H_
 
-#include <fairmq/FairMQDevice.h>
-
 #include <type_traits>
 #include <memory>
 #include <thread>
@@ -26,9 +24,7 @@
 
 #include <boost/dynamic_bitset.hpp>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 template <class F, class ... Args>
@@ -42,26 +38,6 @@ std::thread create_thread_member(const char* name, F&& f, Args&&... args) {
   });
 }
 
-class DataDistDevice : public FairMQDevice {
-
-public:
-
-  void WaitForRunningState() const {
-    while (GetCurrentState() < fair::mq::State::Running) {
-      using namespace std::chrono_literals;
-      std::this_thread::sleep_for(20ms);
-    }
-  }
-
-  bool IsRunningState() const {
-    return (GetCurrentState() == fair::mq::State::Running);
-  }
-
-  bool IsReadyOrRunningState() const {
-    return ((GetCurrentState() == fair::mq::State::Running) || (GetCurrentState() == fair::mq::State::Ready));
-  }
-};
-
 template <
   typename T,
   size_t N = 1024,
@@ -218,7 +194,6 @@ class EventRecorder {
   }
 };
 
-}
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_DATADIST_UTILITIES_H_ */

src/common/fmqtools/CMakeLists.txt

@@ -14,8 +14,9 @@ target_include_directories(fmqtools
     ${CMAKE_CURRENT_SOURCE_DIR}
 )
 
-target_link_libraries(base
+target_link_libraries(fmqtools
   PUBLIC
     FairLogger::FairLogger
     FairMQ::FairMQ
+    base
 )

src/common/fmqtools/FmqUtilities.cxx

@@ -10,3 +10,124 @@
 
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+#include "FmqUtilities.h"
+#include <DataDistLogger.h>
+#include <fairmq/FairMQDevice.h>
+#include <fairmq/DeviceRunner.h>
+
+
+namespace o2::DataDistribution::fmqtools {
+
+static bool checkInfoLoggerOptions() {
+
+  DDDLOG("DataDistLogger: Checking INFOLOGGER_MODE variable");
+
+  const char *cMode = getenv("INFOLOGGER_MODE");
+
+  if (cMode == nullptr) {
+      IDDLOG("DataDistLogger: INFOLOGGER_MODE backend is not set.");
+      return false;
+  }
+
+  const std::string cModeStr = std::string(cMode);
+
+  if (cModeStr.length() == 0) {
+      WDDLOG("DataDistLogger: INFOLOGGER_MODE variable is empty.");
+      return false;
+  }
+
+  if (cModeStr != "infoLoggerD") {
+      EDDLOG("DataDistLogger: INFOLOGGER_MODE mode is not supported "
+      "(only infoLoggerD mode). INFOLOGGER_MODE={}", cModeStr);
+      return false;
+  }
+
+  IDDLOG("DataDistLogger: enabling InfoLogger in infoLoggerD mode.");
+
+  return true;
+}
+
+
+static void handleSeverity(const std::string &pOptKey, const std::string &pOptVal)
+{
+  if (pOptKey == "severity") {
+    // we never allow FairMQ console or file backends!
+    fair::Logger::SetFileSeverity(DataDistSeverity::nolog);
+    fair::Logger::SetConsoleSeverity(fair::Severity::nolog);
+
+    // set the DD log console severity
+    if (fair::Logger::fSeverityMap.count(pOptVal)) {
+      const auto newLevel = fair::Logger::fSeverityMap.at(pOptVal);
+      if (newLevel == fair::Severity::nolog) {
+        DataDistLogger::sConfigSeverity = fair::Severity::fatal;
+      } else {
+        DataDistLogger::sConfigSeverity = newLevel;
+      }
+    }
+
+  } else if (pOptKey == "severity-infologger") {
+    // check the InfoLogger mode. Only infoLoggerD is supported.
+    if(!checkInfoLoggerOptions()) {
+      WDDLOG("DataDistLogger: Invalid INFOLOGGER_MODE. Ignoring severity-infologger={}",
+        pOptVal);
+      DataDistLogger::sInfologgerSeverity = DataDistSeverity::nolog;
+      DataDistLogger::sInfologgerEnabled = false;
+      return;
+    }
+
+    // set the InfoLogger log severity
+    if (fair::Logger::fSeverityMap.count(pOptVal)) {
+      const auto newLevel = fair::Logger::fSeverityMap.at(pOptVal);
+      if (newLevel == fair::Severity::nolog) {
+        DataDistLogger::sInfologgerSeverity = fair::Severity::fatal;
+        DataDistLogger::sInfologgerEnabled = false;
+      } else {
+        DataDistLogger::sInfologgerSeverity = newLevel;
+        DataDistLogger::sInfologgerEnabled = true;
+        impl::DataDistLoggerCtx::InitInfoLogger();
+      }
+    }
+  } else if (pOptKey == "severity-file") {
+      EDDLOG("DataDistLogger: FMQ File logger is not supported.");
+  }
+}
+
+static auto handleRunNumber(const std::string &pOptKey, const std::string &pOptVal)
+{
+  if (pOptKey == "runNumber") {
+    IDDLOG("NEW RUN NUMBER. run_number={}", pOptVal);
+    DataDistLogger::sRunNumberStr = pOptVal;
+    impl::DataDistLoggerCtx::InitInfoLogger();
+  }
+}
+
+
+void HandleFMQOptions(fair::mq::DeviceRunner &pFMQRunner)
+{
+  fair::mq::ProgOptions& lFMQConfig = pFMQRunner.fConfig;
+
+  // disable fairlogger file backend
+  lFMQConfig.SetProperty<std::string>("file-severity", "nolog");
+  lFMQConfig.SetProperty<std::string>("log-to-file", "");
+
+  try {
+    pFMQRunner.UnsubscribeFromConfigChange();
+  } catch(...) { }
+
+  // subscribe to notifications
+  lFMQConfig.Subscribe<std::string>("dd-log-config", handleSeverity);
+  lFMQConfig.Subscribe<std::string>("run-number-config", handleRunNumber);
+
+  // read and apply the current value
+  handleSeverity("severity", lFMQConfig.GetProperty<std::string>("severity"));
+
+  // read and apply the current value
+  handleSeverity("severity-infologger", lFMQConfig.GetProperty<std::string>("severity-infologger"));
+
+  // set a degfault run number
+  handleRunNumber("runNumber", "0");
+}
+
+
+}