Advanced random data producer (#263)

Author: knopers8

Framework/CMakeLists.txt

@@ -57,7 +57,8 @@ add_library(QualityControl
             src/PostProcessingFactory.cxx
             src/PostProcessingConfig.cxx
             src/PostProcessingInterface.cxx
-            src/DummyDatabase.cxx)
+            src/DummyDatabase.cxx
+            src/DataProducer.cxx)
 
 if(ENABLE_MYSQL)
   target_sources(QualityControl PRIVATE src/MySqlDatabase.cxx)

Framework/include/QualityControl/DataProducer.h

@@ -0,0 +1,46 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+///
+/// \file   DataProducer.h
+/// \author Piotr Konopka
+///
+
+#ifndef QUALITYCONTROL_DATAPRODUCER_H
+#define QUALITYCONTROL_DATAPRODUCER_H
+
+#include <Framework/DataProcessorSpec.h>
+
+namespace o2::quality_control::core
+{
+
+/// \brief Returns a random data producer specification which publishes on {"TST", "RAWDATA", <index>}
+///
+/// \param minSize  Minimum size of a message in bytes
+/// \param maxSize  Maximum size of a message in bytes
+/// \param rate     How much messages to produce in one second
+/// \param fill     Should it fill messages with random data
+/// \param index    SubSpecification of the data producer (useful when more than one needed)
+/// \return         A random data producer specification
+framework::DataProcessorSpec getDataProducerSpec(size_t minSize, size_t maxSize, double rate, bool fill = true, size_t index = 0, std::string monitoringUrl = "");
+
+/// \brief Returns an algorithm generating random messages
+///
+/// \param output   Origin, Description and SubSpecification of data to be produced
+/// \param minSize  Minimum size of a message in bytes
+/// \param maxSize  Maximum size of a message in bytes
+/// \param rate     How much messages to produce in one second
+/// \param fill     Should it fill messages with random data
+/// \return         A random data producer algorithm
+framework::AlgorithmSpec getDataProducerAlgorithm(framework::ConcreteDataMatcher output, size_t minSize, size_t maxSize, double rate, bool fill = true, std::string monitoringUrl = "");
+
+} // namespace o2::quality_control::core
+
+#endif //QUALITYCONTROL_DATAPRODUCER_H

Framework/src/DataProducer.cxx

@@ -0,0 +1,93 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+///
+/// \file   DataProducer.cxx
+/// \author Piotr Konopka
+///
+#include "QualityControl/DataProducer.h"
+
+#include <random>
+#include <Common/Timer.h>
+#include <Monitoring/MonitoringFactory.h>
+
+using namespace o2::framework;
+using namespace o2::monitoring;
+
+using SubSpec = o2::header::DataHeader::SubSpecificationType;
+using namespace AliceO2::Common;
+
+namespace o2::quality_control::core
+{
+
+DataProcessorSpec getDataProducerSpec(size_t minSize, size_t maxSize, double rate, bool fill, size_t index, std::string monitoringUrl)
+{
+  return DataProcessorSpec{
+    "producer-" + std::to_string(index),
+    Inputs{},
+    Outputs{
+      { { "out" }, "TST", "RAWDATA", static_cast<SubSpec>(index) } },
+    getDataProducerAlgorithm({ "TST", "RAWDATA", static_cast<SubSpec>(index) }, minSize, maxSize, rate, fill, monitoringUrl)
+  };
+}
+
+framework::AlgorithmSpec
+  getDataProducerAlgorithm(ConcreteDataMatcher output, size_t minSize, size_t maxSize, double rate, bool fill,
+                           std::string monitoringUrl)
+{
+  return AlgorithmSpec{
+    [=](InitContext&) {
+      // this is the initialization code
+      std::default_random_engine generator(time(nullptr));
+      std::shared_ptr<Timer> timer = nullptr;
+
+      int messageCounter = 0;
+      std::shared_ptr<monitoring::Monitoring> collector;
+      if (!monitoringUrl.empty()) {
+        collector = MonitoringFactory::Get(monitoringUrl);
+        collector->enableProcessMonitoring();
+      }
+
+      // after the initialization, we return the processing callback
+      return [=](ProcessingContext& processingContext) mutable {
+        // everything inside this lambda function is invoked in a loop, because it this Data Processor has no inputs
+
+        // setting up the timer
+        if (!timer) {
+          timer = std::make_shared<Timer>();
+          timer->reset(static_cast<int>(1000000.0 / rate));
+        }
+        // keeping the message rate
+        double timeToSleep = timer->getRemainingTime();
+        if (timeToSleep > 0) {
+          usleep(timeToSleep * 1000000.0);
+        }
+        timer->increment();
+
+        // generating data
+        size_t length = minSize + (generator() % (maxSize - minSize));
+        auto data = processingContext.outputs().make<char>({ output.origin, output.description, output.subSpec },
+                                                           length);
+        if (fill) {
+          for (auto&& item : data) {
+            item = static_cast<char>(generator());
+          }
+        }
+
+        if (collector) {
+          collector->send({ messageCounter++, "Data_producer_" + std::to_string(output.subSpec) + "_message_" },
+                          DerivedMetricMode::RATE);
+        }
+      };
+    }
+  };
+}
+
+} // namespace o2::quality_control::core

Framework/src/runBasic.cxx

@@ -62,7 +62,6 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
     ConfigParamSpec{ "no-data-sampling", VariantType::Bool, false, { "Skips data sampling, connects directly the task to the producer." } });
 }
 
-#include <random>
 #include <string>
 
 #include <Framework/runDataProcessing.h>
@@ -71,6 +70,7 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
 #include "QualityControl/InfrastructureGenerator.h"
 #include "QualityControl/runnerUtils.h"
 #include "QualityControl/ExamplePrinterSpec.h"
+#include "QualityControl/DataProducer.h"
 
 std::string getConfigPath(const ConfigContext& config);
 
@@ -79,38 +79,12 @@ using namespace o2::framework;
 using namespace o2::quality_control::checker;
 using namespace std::chrono;
 
-// clang-format off
 WorkflowSpec defineDataProcessing(const ConfigContext& config)
 {
   WorkflowSpec specs;
 
   // The producer to generate some data in the workflow
-  DataProcessorSpec producer{
-    "producer",
-    Inputs{},
-    Outputs{
-      { "TST", "RAWDATA", 0 } },
-    AlgorithmSpec{
-      (AlgorithmSpec::InitCallback)[](InitContext&){
-        // this is the initialization code
-        std::default_random_engine generator(11);
-
-        // after the initialization, we return the processing callback
-        return (AlgorithmSpec::ProcessCallback)[generator](ProcessingContext & processingContext) mutable
-        {
-          // everything inside this lambda function is invoked in a loop, because it this Data Processor has no inputs
-          usleep(100000);
-
-          size_t length = generator() % 10000;
-          auto data = processingContext.outputs().make<char>(Output{ "TST", "RAWDATA" }, length);
-          for (auto&& item : data) {
-            item = static_cast<char>(generator());
-          }
-        };
-      }
-    }
-  };
-
+  DataProcessorSpec producer = getDataProducerSpec(1, 10000, 10);
   specs.push_back(producer);
 
   // Path to the config file
@@ -135,7 +109,6 @@ WorkflowSpec defineDataProcessing(const ConfigContext& config)
 
   return specs;
 }
-// clang-format on
 
 // TODO merge this with the one from runReadout.cxx
 std::string getConfigPath(const ConfigContext& config)

Framework/src/runDataProducer.cxx

@@ -19,50 +19,51 @@
 /// \code{.sh}
 /// o2-qc-run-producer | o2-qc --config json://${QUALITYCONTROL_ROOT}/etc/basic.json
 /// \endcode
+/// Check out the help message to see how to configure data rate and message size.
 ///
 /// If you have glfw installed, you should see a window with the workflow visualization and sub-windows for each Data
 /// Processor where their logs can be seen. The processing will continue until the main window it is closed. Regardless
 /// of glfw being installed or not, in the terminal all the logs will be shown as well.
 
-#include <random>
-#include <Framework/runDataProcessing.h>
+#include <vector>
+#include <Framework/ConfigParamSpec.h>
 
 using namespace o2;
 using namespace o2::framework;
 
-// clang-format off
-WorkflowSpec defineDataProcessing(const ConfigContext&)
+void customize(std::vector<ConfigParamSpec>& workflowOptions)
 {
-  WorkflowSpec specs;
+  workflowOptions.push_back(
+    ConfigParamSpec{ "min-size", VariantType::Int, 1, { "Minimum size in bytes of produced messages." } });
+  workflowOptions.push_back(
+    ConfigParamSpec{ "max-size", VariantType::Int, 10000, { "Maximum size in bytes of produced messages." } });
+  workflowOptions.push_back(
+    ConfigParamSpec{ "empty", VariantType::Bool, false, { "Don't fill messages with random data." } });
+  workflowOptions.push_back(
+    ConfigParamSpec{ "message-rate", VariantType::Double, 10.0, { "Rate of messages per second." } });
+  workflowOptions.push_back(
+    ConfigParamSpec{ "producers", VariantType::Int, 1, { "Number of producers. Each will have unique SubSpec, counting from 0." } });
+  workflowOptions.push_back(
+    ConfigParamSpec{ "monitoring-url", VariantType::String, "", { "URL of the Monitoring backend" } });
+}
 
-  // The producer to generate some data in the workflow
-  DataProcessorSpec producer{
-    "producer",
-    Inputs{},
-    Outputs{
-      { "TST", "RAWDATA", 0 } },
-    AlgorithmSpec{
-      (AlgorithmSpec::InitCallback)[](InitContext&){
-        // this is the initialization code
-        std::default_random_engine generator(11);
+#include <Framework/runDataProcessing.h>
+#include "QualityControl/DataProducer.h"
 
-        // after the initialization, we return the processing callback
-        return (AlgorithmSpec::ProcessCallback)[generator](ProcessingContext & processingContext) mutable
-        {
-          // everything inside this lambda function is invoked in a loop, because it this Data Processor has no inputs
-          usleep(100000);
+using namespace o2::quality_control::core;
 
-          size_t length = generator() % 10000;
-          auto data = processingContext.outputs().make<char>(Output{ "TST", "RAWDATA" }, length);
-          for (auto&& item : data) {
-            item = static_cast<char>(generator());
-          }
-        };
-      }
-    }
-  };
-  specs.push_back(producer);
+WorkflowSpec defineDataProcessing(const ConfigContext& config)
+{
+  size_t minSize = config.options().get<int>("min-size");
+  size_t maxSize = config.options().get<int>("max-size");
+  bool fill = !config.options().get<bool>("empty");
+  double rate = config.options().get<double>("message-rate");
+  size_t producers = config.options().get<int>("producers");
+  std::string monitoringUrl = config.options().get<std::string>("monitoring-url");
 
+  WorkflowSpec specs;
+  for (size_t i = 0; i < producers; i++) {
+    specs.push_back(getDataProducerSpec(minSize, maxSize, rate, fill, i, monitoringUrl));
+  }
   return specs;
-}
-// clang-format on
+}