Use queries for describing desired data (#182)

knopers8 · web-flow · commit 1e3493659d7e · 2019-07-24T16:18:12.000+02:00
* Use queries for describing desired data

* unnecessary namespace name

* update to latest changes on the upstream

* fix run-qc-advanced

* New query format for EMCal

* wildcards for readout data

* trigger the build

* Add query format description in the first docu occurence
diff --git a/Framework/advanced.json b/Framework/advanced.json
@@ -56,19 +56,7 @@
       "id": "tst1",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "sum",
-          "dataOrigin": "TST",
-          "dataDescription": "SUM"
-        },
-        {
-          "binding": "param",
-          "dataOrigin": "TST",
-          "dataDescription": "PARAM"
-        }
-      ],
-      "subSpec": "2",
+      "query" : "sum:TST/SUM/2;param:TST/PARAM/2",
       "samplingConditions": [
         {
           "condition": "random",
@@ -82,14 +70,7 @@
       "id": "tst2",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "data",
-          "dataOrigin": "TST",
-          "dataDescription": "DATA"
-        }
-      ],
-      "subSpec": "*",
+      "query" : "data:TST/RAWDATA",
       "samplingConditions": [
         {
           "condition": "payloadSize",
diff --git a/Framework/basic-no-sampling.json b/Framework/basic-no-sampling.json
@@ -28,10 +28,7 @@
         "maxNumberCycles": "-1",
         "dataSource": {
           "type": "direct",
-          "binding": "its-rawdata",
-          "dataOrigin": "ITS",
-          "dataDescription": "RAWDATA",
-          "subSpec": "0"
+          "query" : "its-rawdata:ITS/RAWDATA/0"
         },
         "taskParameters": {
           "nothing": "rien"
diff --git a/Framework/basic.json b/Framework/basic.json
@@ -43,14 +43,7 @@
       "id": "its-raw",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "random",
-          "dataOrigin": "ITS",
-          "dataDescription": "RAWDATA"
-        }
-      ],
-      "subSpec": "0",
+      "query" : "random:ITS/RAWDATA/0",
       "samplingConditions": [
         {
           "condition": "random",
diff --git a/Framework/example-default.json b/Framework/example-default.json
@@ -65,14 +65,8 @@
     {
       "id": "ex1",
       "active": "true",
-      "dataHeaders": [
-        {
-          "binding": "data",
-          "dataOrigin": "TST",
-          "dataDescription": "DATA"
-        }
-      ],
-      "subSpec": "0",
+      "query_comment" : "query is in the format of binding1:origin1/description1/subSpec1;binding2:origin2/description2/subSpec2;...",
+      "query" : "data:TST/DATA/0",
       "samplingConditions": [
         {
           "condition": "custom",
@@ -85,14 +79,7 @@
     {
       "id": "mftclusters",
       "active": "true",
-      "dataHeaders": [
-        {
-          "binding": "mft-clusters",
-          "dataOrigin": "MFT",
-          "dataDescription": "CLUSTERS"
-        }
-      ],
-      "subSpec": "0",
+      "query" : "mft-clusters:MFT/CLUSTERS/0",
       "samplingConditions": [
         {
           "condition": "payloadSize",
diff --git a/Framework/readout-no-sampling.json b/Framework/readout-no-sampling.json
@@ -28,10 +28,7 @@
         "maxNumberCycles": "-1",
         "dataSource": {
           "type": "direct",
-          "binding": "readout",
-          "dataOrigin": "ITS",
-          "dataDescription": "RAWDATA",
-          "subSpec": "0"
+          "query" : "readout:ROUT/RAWDATA"
         },
         "location": "remote"
       }
diff --git a/Framework/readout.json b/Framework/readout.json
@@ -39,14 +39,7 @@
       "id": "readout",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "readout",
-          "dataOrigin": "ITS",
-          "dataDescription": "RAWDATA"
-        }
-      ],
-      "subSpec": "0",
+      "query" : "readout:ROUT/RAWDATA",
       "samplingConditions": [
         {
           "condition": "random",
diff --git a/Framework/readoutForDataDump.json b/Framework/readoutForDataDump.json
@@ -4,14 +4,7 @@
       "id": "readout",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "readout",
-          "dataOrigin": "R/O",
-          "dataDescription": "RAWDATA"
-        }
-      ],
-      "subSpec": "0",
+      "query" : "readout:RO/RAWDATA/0",
       "samplingConditions": [],
       "blocking": "false",
       "fairMQOutput": "name=fairReadoutRawOut,type=pub,method=bind,address=tcp://127.0.0.1:26525,rateLogging=1"
diff --git a/Framework/src/TaskRunner.cxx b/Framework/src/TaskRunner.cxx
@@ -27,6 +27,7 @@
 #include <Framework/CallbackService.h>
 #include <Framework/TimesliceIndex.h>
 #include <Framework/DataSpecUtils.h>
+#include <Framework/DataDescriptorQueryBuilder.h>
 #include <Headers/DataHeader.h>
 #include <Monitoring/MonitoringFactory.h>
 #include "QualityControl/QcInfoLogger.h"
@@ -274,22 +275,8 @@ void TaskRunner::populateConfig(std::string taskName)
       LOG(INFO) << "policyName : " << policyName;
       mInputSpecs = DataSampling::InputSpecsForPolicy(config, policyName);
     } else if (type == "direct") {
-
-      auto subSpecString = dataSourceTree.get<std::string>("subSpec");
-      auto subSpec = std::strtoull(subSpecString.c_str(), nullptr, 10);
-
-      header::DataOrigin origin;
-      header::DataDescription description;
-      origin.runtimeInit(dataSourceTree.get<std::string>("dataOrigin").c_str());
-      description.runtimeInit(dataSourceTree.get<std::string>("dataDescription").c_str());
-
-      mInputSpecs.push_back(
-        InputSpec{
-          dataSourceTree.get<std::string>("binding"),
-          origin,
-          description,
-          static_cast<framework::DataAllocator::SubSpecificationType>(subSpec) });
-
+      auto inputsQuery = dataSourceTree.get<std::string>("query");
+      mInputSpecs = DataDescriptorQueryBuilder::parse(inputsQuery.c_str());
     } else {
       std::string message = std::string("Configuration error : dataSource type unknown : ") + type; // TODO pass this message to the exception
       BOOST_THROW_EXCEPTION(AliceO2::Common::FatalException() << AliceO2::Common::errinfo_details(message));
diff --git a/Framework/src/runAdvanced.cxx b/Framework/src/runAdvanced.cxx
@@ -136,12 +136,17 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
     DataSampling::GenerateInfrastructure(localTopology, qcConfigurationSource);
     // a fix to make the topologies work when merged together
     localTopology.back().name += std::to_string(i);
+    if (i != 2) {
+      localTopology.back().inputs = { localTopology.back().inputs.back() };
+      localTopology.back().outputs = { localTopology.back().outputs.back() };
+    }
+    DataSpecUtils::updateMatchingSubspec(localTopology.back().inputs.back(), i);
+    DataSpecUtils::updateMatchingSubspec(localTopology.back().outputs.back(), i);
 
     std::string host = "o2flptst" + std::to_string(i);
     quality_control::generateLocalInfrastructure(localTopology, qcConfigurationSource, host);
     // a fix to make the topologies work when merged together
     localTopology.back().name += std::to_string(i);
-    // temporary fix, which shouldn't be necessary when data sampling uses matchers
     DataSpecUtils::updateMatchingSubspec(localTopology.back().inputs[0], i);
     DataSpecUtils::updateMatchingSubspec(localTopology.back().inputs[1], i);
 
diff --git a/Framework/src/runReadout.cxx b/Framework/src/runReadout.cxx
@@ -78,9 +78,9 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
   WorkflowSpec specs{
     specifyExternalFairMQDeviceProxy(
       "readout-proxy",
-      Outputs{{"ITS", "RAWDATA"}},
+      Outputs{ { { "readout" }, { "ROUT", "RAWDATA" } } },
       "type=sub,method=connect,address=ipc:///tmp/readout-pipe-1,rateLogging=1",
-      dataSamplingReadoutAdapter({"ITS", "RAWDATA"}))
+      dataSamplingReadoutAdapter({ { "readout" }, { "ROUT", "RAWDATA" } }))
   };
 
   // Path to the config file
diff --git a/Framework/src/runReadoutForDataDump.cxx b/Framework/src/runReadoutForDataDump.cxx
@@ -55,9 +55,9 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
   WorkflowSpec specs{
     specifyExternalFairMQDeviceProxy(
       "readout-proxy",
-      Outputs{ { "R/O", "RAWDATA" } },
+      Outputs{ { "RO", "RAWDATA" } },
       "type=sub,method=connect,address=ipc:///tmp/readout-pipe-1,rateLogging=1",
-      dataSamplingReadoutAdapter({ "R/O", "RAWDATA" }))
+      dataSamplingReadoutAdapter({ "RO", "RAWDATA" }))
   };
 
   const std::string qcConfigurationSource =
diff --git a/Framework/test/testSharedConfig.json b/Framework/test/testSharedConfig.json
@@ -60,14 +60,7 @@
       "id": "tpcclust",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "clusters",
-          "dataOrigin": "TPC",
-          "dataDescription": "CLUSTERS"
-        }
-      ],
-      "subSpec": "0",
+      "query": "clusters:TPC/CLUSTERS/0",
       "samplingConditions": [
         {
           "condition": "random",
diff --git a/Framework/test/testWorkflow.json b/Framework/test/testWorkflow.json
@@ -35,13 +35,7 @@
       "id": "test-policy",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "tst-data",
-          "dataOrigin": "TST",
-          "dataDescription": "DATA"
-        }
-      ],
+      "query": "tst-data:TST/DATA/0",
       "subSpec": "0",
       "samplingConditions": [],
       "blocking": "false"
diff --git a/Modules/EMCAL/etc/digits.json b/Modules/EMCAL/etc/digits.json
@@ -28,10 +28,7 @@
           "maxNumberCycles": "-1",
           "dataSource": {
             "type": "direct",
-            "binding": "emcal-digits",
-            "dataOrigin": "EMC",
-            "dataDescription": "DIGITS",
-            "subSpec": "0"
+            "query": "emcal-digits:EMC/DIGITS/0"
           },
           "taskParameters": {
             "nothing": "rien"
diff --git a/Modules/Skeleton/src/SkeletonTask.cxx b/Modules/Skeleton/src/SkeletonTask.cxx
@@ -61,11 +61,8 @@ void SkeletonTask::startOfCycle()
 void SkeletonTask::monitorData(o2::framework::ProcessingContext& ctx)
 {
   // In this function you can access data inputs specified in the JSON config file, for example:
-  //  {
-  //    "binding": "random",
-  //    "dataOrigin": "ITS",
-  //    "dataDescription": "RAWDATA"
-  //  }
+  //   "query": "random:ITS/RAWDATA/0"
+  // which is correspondingly <binding>:<dataOrigin>/<dataDescription>/<subSpecification
 
   // Use Framework/DataRefUtils.h or Framework/InputRecord.h to access and unpack inputs (both are documented)
   // One can find additional examples at:
diff --git a/Modules/TOF/src/TOFTask.cxx b/Modules/TOF/src/TOFTask.cxx
@@ -306,11 +306,8 @@ void TOFTask::startOfCycle()
 void TOFTask::monitorData(o2::framework::ProcessingContext& ctx)
 {
   // In this function you can access data inputs specified in the JSON config file, for example:
-  //  {
-  //    "binding": "random",
-  //    "dataOrigin": "ITS",
-  //    "dataDescription": "RAWDATA"
-  //  }
+  //   "query": "random:ITS/RAWDATA/0"
+  // which is correspondingly <binding>:<dataOrigin>/<dataDescription>/<subSpecification
 
   // Use Framework/DataRefUtils.h or Framework/InputRecord.h to access and unpack inputs (both are documented)
   // One can find additional examples at:
diff --git a/Modules/TOF/tof.json b/Modules/TOF/tof.json
@@ -43,14 +43,7 @@
       "id": "its-raw",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "random",
-          "dataOrigin": "ITS",
-          "dataDescription": "RAWDATA"
-        }
-      ],
-      "subSpec": "0",
+      "query" : "random:ITS/RAWDATA/0",
       "samplingConditions": [
         {
           "condition": "random",
diff --git a/doc/ModulesDevelopment.md b/doc/ModulesDevelopment.md
@@ -67,14 +67,8 @@ Data Sampling is used by Quality Control to feed the tasks with data. Below we p
       "id": "its-raw",
       "active": "true",
       "machines": [],
-      "dataHeaders": [
-        {
-          "binding": "raw",
-          "dataOrigin": "ITS",
-          "dataDescription": "RAWDATA"
-        }
-      ],
-      "subSpec": "0",
+      "query_comment" : "query is in the format of binding1:origin1/description1/subSpec1[;binding2:...]",
+      "query": "raw:ITS/RAWDATA/0",
       "samplingConditions": [
         {
           "condition": "random",
@@ -109,10 +103,8 @@ In case one needs to sample at a very high rate, or even monitor 100% of the dat
         ...
         "dataSource": {
           "type": "direct",
-          "binding": "its-rawdata",
-          "dataOrigin": "ITS",
-          "dataDescription": "RAWDATA",
-          "subSpec": "0"
+          "query_comment" : "query is in the format of binding1:origin1/description1/subSpec1[;binding2:...]",
+          "query" : "its-raw-data:ITS/RAWDATA/0"
         },
         ...
       }
