Readme and cleanup of data sampling

Author: knopers8

Framework/Core/CMakeLists.txt

@@ -44,6 +44,7 @@ set(SRCS
     src/DeviceSpec.cxx
     src/DeviceSpecHelpers.cxx
     src/DDSConfigHelpers.cxx
+    src/Dispatcher.cxx
     src/DriverControl.cxx
     src/DriverInfo.cxx
     src/FairOptionsRetriever.cxx
@@ -60,7 +61,6 @@ set(SRCS
     src/TextControlService.cxx
     src/TableBuilder.cxx
     src/TableConsumer.cxx
-    src/WorkflowDispatcher.cxx
     src/WorkflowHelpers.cxx
     src/WorkflowSpec.cxx
     src/runDataProcessing.cxx
@@ -145,10 +145,10 @@ set(HEADERS
       include/Framework/DataSamplingConditionFactory.h
       include/Framework/DataSamplingReadoutAdapter.h
       include/Framework/DataSamplingPolicy.h
+      include/Framework/Dispatcher.h
       include/Framework/DPLBoostSerializer.h
       include/Framework/TableBuilder.h
       include/Framework/FairMQResizableBuffer.h
-      include/Framework/WorkflowDispatcher.h
       src/ComputingResource.h
       src/DDSConfigHelpers.h
       src/DeviceSpecHelpers.h
@@ -200,7 +200,7 @@ O2_GENERATE_EXECUTABLE(
 target_compile_options(Framework PUBLIC -O0 -g -fno-omit-frame-pointer)
 target_compile_options(test_SimpleDataProcessingDevice01 PUBLIC -O0 -g -fno-omit-frame-pointer)
 
-Install(FILES test/test_DataSamplingDPL.json test/test_DataSamplingFairMQ.json DESTINATION share/tests/)
+Install(FILES test/test_DataSampling.json DESTINATION share/tests/)
 
 set(TEST_SRCS
       test/test_AlgorithmSpec.cxx
@@ -247,7 +247,6 @@ set(TEST_SRCS
       test/test_TimeParallelPipelining.cxx
       test/test_TypeTraits.cxx
       test/test_Variants.cxx
-#      test/test_WorkflowDispatcher.cxx
       test/test_WorkflowHelpers.cxx
    )
 

Framework/Core/README.md

@@ -437,7 +437,102 @@ There is also a few demonstrator available in particular:
 - [MillWheel: Fault-Tolerant Stream Processing at Internet Scale](https://research.google.com/pubs/pub41378.html) : paper about Google previous generation system for stream processing
 - [Concord](http://concord.io) : Similar (to the above) stream processing solution, OpenSource.
 
+## Data Sampling
 
+Data Sampling provides possibility to sample data in DPL workflows, basing on certain conditions ( 5% randomly, when payload is greater than 4234 bytes, etc.). The job of passing the right data is done by a data processor called `Dispatcher`. A desired data stream is specified in form of Data Sampling Policies, configured by JSON structures (example below).
+```
+{
+  "id": "policy_example1",              # name of the policy
+  "active": "false",                    # activation flag
+  "machines": [                         # list of machines where the policy should be run (now ignored)
+    "aido2flp1",
+    "aido2flp2"
+  ],
+  "dataHeaders": [                      # list of data that should be sampled
+    {
+      "binding": "clusters",            # binding of the data in InputRecord
+      "dataOrigin": "TPC",              # data origin in DataHeader
+      "dataDescription": "CLUSTERS"     # data description in DataHeader
+    },
+    {
+      "binding": "tracks",
+      "dataOrigin": "TPC",
+      "dataDescription": "TRACKS"
+    }
+  ],
+  "subSpec": "0",                       # subspecification in DataHeader, use -1 for all
+  "samplingConditions": [               # list of sampling conditions
+    {
+      "condition": "random",            # condition type
+      "fraction": "0.1",                # condition-dependent parameter: fraction of data to sample
+      "seed": "2112"                    # condition-dependent parameter: seed of PRNG
+    }
+  ],
+  "blocking": "false"                   # should the dispatcher block the main data flow? (now ignored)
+}
+```
+
+### Usage
+
+To use Data Sampling in a DPL workflow insert following lines to your code:
+```
+#include "Framework/DataSampling.h"
+using namespace o2::framework;
+void customize(std::vector<CompletionPolicy>& policies)
+{
+  DataSampling::CustomizeInfrastructure(policies);
+}
+
+void customize(std::vector<ChannelConfigurationPolicy>& policies)
+{
+  DataSampling::CustomizeInfrastructure(policies);
+}
+
+#include "Framework/runDataProcessing.h"
+
+std::vector<DataProcessorSpec> defineDataProcessing(ConfigContext &ctx)
+{
+
+ WorkflowSpec workflow;
+// <declaration of other DPL processors>
+
+ DataSampling::GenerateInfrastructure(workflow, "json:///absolute/path/to/config/file.json");
+
+ return workflow;
+}
+```
+
+Sampled data can be subscribed to by adding `InputSpecs` provided by `std::vector<InputSpec> DataSampling::InputSpecsForPolicy(const std::string& policiesSource, const std::string& policyName)` to a chosen data processor. Then, they can be accessed by the bindings specified in the configuration file.
+
+[dataSamplingPodAndRoot](https://github.com/AliceO2Group/AliceO2/blob/dev/Framework/TestWorkflows/src/dataSamplingPodAndRoot.cxx) workflow can serve as usage example.
+
+## Data Sampling Conditions
+
+The following sampling conditions are available. When more than one is used, a positive decision is taken when all the conditions are fulfilled.
+- **DataSamplingConditionRandom** - pseudo-randomly accepts specified fraction of incoming messages.
+```
+{
+  "condition": "random",
+  "fraction": "0.1",
+  "seed": "22222"
+}
+```
+- **DataSamplingConditionNConsecutive** - approves n consecutive samples in defined cycle. It assumes that timesliceID always increments by one.
+```
+{
+  "condition": "nConsecutive",
+  "samplesNumber": "3",
+  "cycleSize": "100"
+}
+```
+- **DataSamplingConditionPayloadSize** - approves messages having payload size within specified boundaries.
+```
+{
+  "condition": "payloadSize",
+  "lowerLimit": "300",
+  "upperLimit": "500"
+}
+```
 ## Document history
 
 * v0.9: proposal for approval at the O2 TB - 19th June 2018

Framework/Core/include/Framework/DataSampling.h

@@ -16,12 +16,14 @@
 ///
 /// \author Piotr Konopka, piotr.jan.konopka@cern.ch
 
-#include <string>
-
 #include "Framework/WorkflowSpec.h"
 #include "Framework/CompletionPolicy.h"
-#include "Framework/DataSamplingPolicy.h"
 #include "Framework/ChannelConfigurationPolicy.h"
+#include "Framework/InputSpec.h"
+
+#include <Configuration/ConfigurationInterface.h>
+
+#include <string>
 
 namespace o2
 {
@@ -60,9 +62,6 @@ namespace framework
 /// }
 /// \endcode
 
-//todo: update docu
-//todo: clean header mess
-
 class DataSampling
 {
  public:

Framework/Core/include/Framework/DataSamplingCondition.h

@@ -17,6 +17,7 @@
 /// \author Piotr Konopka, piotr.jan.konopka@cern.ch
 
 #include "Framework/DataRef.h"
+
 #include <boost/property_tree/ptree.hpp>
 
 namespace o2

Framework/Core/include/Framework/DataSamplingConfig.h

@@ -16,15 +16,14 @@
 ///
 /// \author Piotr Konopka, piotr.jan.konopka@cern.ch
 
-#include <mutex>
-#include <boost/property_tree/ptree.hpp>
-
 #include "Headers/DataHeader.h"
 #include "Framework/InputSpec.h"
 #include "Framework/Output.h"
 #include "Framework/OutputSpec.h"
 #include "Framework/DataSamplingCondition.h"
 
+#include <boost/property_tree/ptree.hpp>
+
 namespace o2
 {
 namespace framework
@@ -37,12 +36,13 @@ namespace framework
 class DataSamplingPolicy
 {
  private:
+  // todo: see if dpl matchers can be used here instead of this strange construction
   struct inputSpecHasher {
     size_t operator()(const InputSpec& i) const
     {
       return (static_cast<size_t>(i.description.itg[0]) << 32 |
-             static_cast<size_t>(i.description.itg[1])) ^
-               static_cast<size_t>(i.origin.itg[0]);
+              static_cast<size_t>(i.description.itg[1])) ^
+             static_cast<size_t>(i.origin.itg[0]);
     }
   };
   struct inputSpecEqual {
@@ -76,7 +76,6 @@ class DataSamplingPolicy
   const Output prepareOutput(const InputSpec&) const;
 
   const std::string& getName() const;
-  //  const std::vector<InputSpec>& getInputs() const;
   const PathMap& getPathMap() const;
   const header::DataHeader::SubSpecificationType getSubSpec() const;
   // optional fairmq channel to send stuff outside of DPL
@@ -92,8 +91,6 @@ class DataSamplingPolicy
   header::DataHeader::SubSpecificationType mSubSpec;
   std::vector<std::unique_ptr<DataSamplingCondition>> mConditions;
   std::string mFairMQOutputChannel;
-
-  std::mutex mDecisionMutex;
 };
 
 } // namespace framework

Framework/Core/include/Framework/DataSamplingPolicy.h

@@ -8,13 +8,13 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
-/// \file WorkflowDispatcher.h
-/// \brief Declaration of WorkflowDispatcher for O2 Data Sampling
+/// \file Dispatcher.h
+/// \brief Declaration of Dispatcher for O2 Data Sampling
 ///
 /// \author Piotr Konopka, piotr.jan.konopka@cern.ch
 
-#ifndef ALICEO2_WORKFLOWDISPATCHER_H
-#define ALICEO2_WORKFLOWDISPATCHER_H
+#ifndef ALICEO2_DISPATCHER_H
+#define ALICEO2_DISPATCHER_H
 
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/DataSamplingPolicy.h"
@@ -25,13 +25,13 @@ namespace o2
 namespace framework
 {
 
-class WorkflowDispatcher : public Task
+class Dispatcher : public Task
 {
  public:
   /// \brief Constructor
-  WorkflowDispatcher(const std::string name = std::string(), const std::string reconfSource = std::string());
+  Dispatcher(const std::string name, const std::string reconfigurationSource);
   /// \brief Destructor
-  ~WorkflowDispatcher();
+  ~Dispatcher();
 
   /// \brief Dispatcher init callback
   void init(InitContext& ctx) override;
@@ -60,4 +60,4 @@ class WorkflowDispatcher : public Task
 } // namespace framework
 } // namespace o2
 
-#endif //ALICEO2_WORKFLOWDISPATCHER_H
+#endif //ALICEO2_DISPATCHER_H