[QC-114] Use a container to send all MOs at once (#93)

knopers8 · web-flow · commit ddadadb8a4a8 · 2018-12-10T09:54:07.000+01:00
* Use container to send all MOs at once

* Send TObjArray instead of MOs in Checker
diff --git a/Framework/include/QualityControl/Checker.h b/Framework/include/QualityControl/Checker.h
@@ -82,7 +82,7 @@ class Checker : public framework::Task
   /**
    * \brief Send the MonitorObject on FairMQ to whoever is listening.
    */
-  void send(std::shared_ptr<MonitorObject> mo, framework::DataAllocator& allocator);
+  void send(std::unique_ptr<TObjArray>& mo, framework::DataAllocator& allocator);
 
   /**
    * \brief Load a library.
diff --git a/Framework/include/QualityControl/HistoMerger.h b/Framework/include/QualityControl/HistoMerger.h
@@ -56,7 +56,7 @@ class HistoMerger : public framework::Task
  private:
   // General state
   std::string mMergerName;
-  std::shared_ptr<MonitorObject> mMonitorObject;
+  TObjArray mMergedArray;
   AliceO2::Common::Timer mPublicationTimer;
 
   // DPL
diff --git a/Framework/include/QualityControl/ObjectsManager.h b/Framework/include/QualityControl/ObjectsManager.h
@@ -9,6 +9,7 @@
 #include "QualityControl/MonitorObject.h"
 #include "QualityControl/Quality.h"
 #include "QualityControl/TaskConfig.h"
+#include <TObjArray.h>
 #include <TObjString.h>
 #include <boost/concept_check.hpp>
 #include <string>
@@ -68,22 +69,15 @@ class ObjectsManager
 
   TObject* getObject(std::string objectName);
 
-  typedef typename std::map<std::string, quality_control::core::MonitorObject*>::iterator iterator;
-  typedef typename std::map<std::string, quality_control::core::MonitorObject*>::const_iterator const_iterator;
-
-  const_iterator begin() const { return mMonitorObjects.begin(); }
-  const_iterator end() const { return mMonitorObjects.end(); }
-
-  iterator begin() { return mMonitorObjects.begin(); }
-  iterator end() { return mMonitorObjects.end(); }
+  TObjArray* getNonOwningArray() const { return new TObjArray(mMonitorObjects); };
 
   std::string getObjectsListString() { return mObjectsList.GetString().Data(); }
 
  private:
   void UpdateIndex(const std::string& nonEmptyName);
 
  private:
-  std::map<std::string /*object name*/, quality_control::core::MonitorObject* /* object */> mMonitorObjects;
+  TObjArray mMonitorObjects;
   std::string mTaskName;
   // todo make it a vector of string when support added
   TObjString mObjectsList; // the list of objects we publish. (comma separated)
diff --git a/Framework/src/Checker.cxx b/Framework/src/Checker.cxx
@@ -14,6 +14,7 @@
 #include <Common/Exceptions.h>
 #include <Configuration/ConfigurationFactory.h>
 #include <Framework/DataRefUtils.h>
+#include <TMap.h>
 // QC
 #include "QualityControl/DatabaseFactory.h"
 #include "QualityControl/TaskRunner.h"
@@ -101,20 +102,26 @@ void Checker::run(framework::ProcessingContext& ctx)
     startFirstObject = system_clock::now();
   }
 
-  for (auto&& input : ctx.inputs()) {
-    // will that work properly with shmem?
-    std::shared_ptr<MonitorObject> mo{ std::move(framework::DataRefUtils::as<MonitorObject>(input)) };
+  std::shared_ptr<TObjArray> moArray{ std::move(framework::DataRefUtils::as<TObjArray>(*ctx.inputs().begin())) };
+  moArray->SetOwner(false);
+  auto checkedMoArray = std::make_unique<TObjArray>();
+  checkedMoArray->SetOwner();
 
+  for (const auto& to : *moArray) {
+    std::shared_ptr<MonitorObject> mo{dynamic_cast<MonitorObject*>(to)};
+    moArray->RemoveFirst();
     if (mo) {
       check(mo);
       store(mo);
-      send(mo, ctx.outputs());
       mTotalNumberHistosReceived++;
+      checkedMoArray->Add(new MonitorObject(*mo));
     } else {
       mLogger << "the mo is null" << AliceO2::InfoLogger::InfoLogger::endm;
     }
   }
 
+  send(checkedMoArray, ctx.outputs());
+
   // monitoring
   endLastObject = system_clock::now();
 
@@ -169,13 +176,12 @@ void Checker::store(std::shared_ptr<MonitorObject> mo)
   }
 }
 
-void Checker::send(std::shared_ptr<MonitorObject> mo, framework::DataAllocator& allocator)
+void Checker::send(std::unique_ptr<TObjArray>& moArray, framework::DataAllocator& allocator)
 {
-  mLogger << "Sending \"" << mo->getName() << "\"" << AliceO2::InfoLogger::InfoLogger::endm;
-
-  // todo: consider adopting
-  allocator.snapshot<MonitorObject>(
-    framework::Output{ mOutputSpec.origin, mOutputSpec.description, mOutputSpec.subSpec, mOutputSpec.lifetime }, *mo);
+  mLogger << "Sending Monitor Object array with " << moArray->GetEntries() << " objects inside." << AliceO2::InfoLogger::InfoLogger::endm;
+  
+  allocator.adopt(
+    framework::Output{ mOutputSpec.origin, mOutputSpec.description, mOutputSpec.subSpec, mOutputSpec.lifetime }, moArray.release());
 }
 
 void Checker::loadLibrary(const std::string libraryName)
diff --git a/Framework/src/HistoMerger.cxx b/Framework/src/HistoMerger.cxx
@@ -6,6 +6,7 @@
 #include "QualityControl/HistoMerger.h"
 
 #include <Framework/DataRefUtils.h>
+#include <TObjArray.h>
 
 namespace o2
 {
@@ -22,33 +23,42 @@ HistoMerger::HistoMerger(std::string mergerName, double publicationPeriodSeconds
   : mMergerName(mergerName), mOutputSpec{ header::gDataOriginInvalid, header::gDataDescriptionInvalid }
 {
   mPublicationTimer.reset(static_cast<int>(publicationPeriodSeconds * 1000000));
+  mMergedArray.SetOwner(true);
 }
 
 HistoMerger::~HistoMerger() {}
 
-void HistoMerger::init(framework::InitContext& ctx) { mMonitorObject.reset(); }
+void HistoMerger::init(framework::InitContext&) { mMergedArray.Clear(); }
 
 void HistoMerger::run(framework::ProcessingContext& ctx)
 {
   for (const auto& input : ctx.inputs()) {
-    if (input.header != nullptr && input.spec != nullptr &&
-        std::strstr(DataRefUtils::as<MonitorObject>(input)->getObject()->ClassName(), "TH1") != nullptr) {
+    if (input.header != nullptr && input.spec != nullptr) {
+      std::unique_ptr<TObjArray> moArray = DataRefUtils::as<TObjArray>(input);
 
-      if (!mMonitorObject) {
-        mMonitorObject.reset(DataRefUtils::as<MonitorObject>(input).release());
+      if (mMergedArray.IsEmpty()) {
+        mMergedArray = *moArray.release();
       } else {
-        TH1* h = dynamic_cast<TH1*>(mMonitorObject->getObject());
-        const TH1* hUpdate = dynamic_cast<TH1*>(DataRefUtils::as<MonitorObject>(input)->getObject());
-        h->Add(hUpdate);
+        if (mMergedArray.GetSize() != moArray->GetSize()) {
+          LOG(ERROR) << "array don't match in size, " << mMergedArray.GetSize() << " vs " << moArray->GetSize();
+          return;
+        }
+
+        for (int i = 0; i < mMergedArray.GetEntries(); i++) {
+          MonitorObject* mo = dynamic_cast<MonitorObject*>((*moArray)[i]);
+          if (mo && std::strstr(mo->getObject()->ClassName(), "TH1") != nullptr) {
+            TH1* h = dynamic_cast<TH1*>(dynamic_cast<MonitorObject*>(mMergedArray[i])->getObject());
+            const TH1* hUpdate = dynamic_cast<TH1*>(mo->getObject());
+            h->Add(hUpdate);
+          }
+        }
       }
     }
   }
   if (mPublicationTimer.isTimeout()) {
-    if (mMonitorObject) {
-      ctx.outputs().snapshot<MonitorObject>(Output{ mOutputSpec.origin, mOutputSpec.description, mOutputSpec.subSpec },
-                                            *mMonitorObject);
+    if (!mMergedArray.IsEmpty()) {
+      ctx.outputs().snapshot(Output{ mOutputSpec.origin, mOutputSpec.description, mOutputSpec.subSpec }, mMergedArray);
     }
-
     // avoid publishing mo many times consecutively because of too long initial waiting time
     do {
       mPublicationTimer.increment();
diff --git a/Framework/src/ObjectsManager.cxx b/Framework/src/ObjectsManager.cxx
@@ -31,22 +31,19 @@ namespace core
 ObjectsManager::ObjectsManager(TaskConfig& taskConfig) : mTaskName(taskConfig.taskName)
 {
   startPublishing(&mObjectsList, MonitorObject::SYSTEM_OBJECT_PUBLICATION_LIST);
+  mMonitorObjects.SetOwner(true);
 }
 
 ObjectsManager::~ObjectsManager()
 {
-  for (auto& mMonitorObject : mMonitorObjects) {
-    delete mMonitorObject.second;
-  }
-  mMonitorObjects.clear();
 }
 
 void ObjectsManager::startPublishing(TObject* object, std::string objectName)
 {
   std::string nonEmptyName = objectName.empty() ? object->GetName() : objectName;
   auto* newObject = new MonitorObject(nonEmptyName, object, mTaskName);
   newObject->setIsOwner(false);
-  mMonitorObjects[nonEmptyName] = newObject;
+  mMonitorObjects.Add(newObject);
 
   // update index
   if (objectName != MonitorObject::SYSTEM_OBJECT_PUBLICATION_LIST) {
@@ -83,8 +80,10 @@ void ObjectsManager::addCheck(const std::string& objectName, const std::string&
 
 MonitorObject* ObjectsManager::getMonitorObject(std::string objectName)
 {
-  if (mMonitorObjects.count(objectName) > 0) {
-    return mMonitorObjects[objectName];
+  TObject* mo = mMonitorObjects.FindObject(objectName.c_str());
+
+  if (mo) {
+    return dynamic_cast<MonitorObject*>(mo);
   } else {
     BOOST_THROW_EXCEPTION(ObjectNotFoundError() << errinfo_object_name(objectName));
   }
diff --git a/Framework/src/TaskRunner.cxx b/Framework/src/TaskRunner.cxx
@@ -47,7 +47,7 @@ TaskRunner::TaskRunner(std::string taskName, std::string configurationSource, si
     mLastNumberObjects(0),
     mCycleOn(false),
     mCycleNumber(0),
-    mMonitorObjectsSpec(createTaskDataOrigin(), createTaskDataDescription(taskName), id),
+    mMonitorObjectsSpec({"mo"}, createTaskDataOrigin(), createTaskDataDescription(taskName), id),
     mResetAfterPublish(false),
     mTask(nullptr)
 {
@@ -269,20 +269,15 @@ void TaskRunner::finishCycle(DataAllocator& outputs)
 
 unsigned long TaskRunner::publish(DataAllocator& outputs)
 {
-  unsigned int sentMessages = 0;
-
-  for (auto& pair : *mObjectsManager) {
-
-    auto* mo = pair.second;
-    outputs.snapshot<MonitorObject>(Output{ mMonitorObjectsSpec.origin, mMonitorObjectsSpec.description,
-                                            mMonitorObjectsSpec.subSpec, mMonitorObjectsSpec.lifetime },
-                                    *mo);
-
-    QcInfoLogger::GetInstance() << "Sending \"" << mo->getName() << "\"" << AliceO2::InfoLogger::InfoLogger::endm;
-    sentMessages++;
-  }
-
-  return sentMessages;
+  outputs.adopt(
+    Output{ mMonitorObjectsSpec.origin,
+            mMonitorObjectsSpec.description,
+            mMonitorObjectsSpec.subSpec,
+            mMonitorObjectsSpec.lifetime },
+    dynamic_cast<TObject*>(mObjectsManager->getNonOwningArray())
+  );
+
+  return 1;
 }
 
 void TaskRunner::CustomCleanupTMessage(void* data, void* object) { delete (TMessage*)object; }
diff --git a/Framework/src/runBasic.cxx b/Framework/src/runBasic.cxx
@@ -108,15 +108,19 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
 
         return (AlgorithmSpec::ProcessCallback) [](ProcessingContext& processingContext) mutable {
           LOG(INFO) << "printer invoked";
-          auto mo = processingContext.inputs().get<MonitorObject*>("checked-mo").get();
+          std::shared_ptr<TObjArray> moArray{ std::move(DataRefUtils::as<TObjArray>(*processingContext.inputs().begin())) };
 
-          if (mo->getName() == "example") {
-            auto* g = dynamic_cast<TH1F*>(mo->getObject());
-            std::string bins = "BINS:";
-            for (int i = 0; i < g->GetNbinsX(); i++) {
-              bins += " " + std::to_string((int) g->GetBinContent(i));
+          for (const auto& to : *moArray) {
+            MonitorObject* mo = dynamic_cast<MonitorObject*>(to);
+
+            if (mo->getName() == "example") {
+              auto* g = dynamic_cast<TH1F*>(mo->getObject());
+              std::string bins = "BINS:";
+              for (int i = 0; i < g->GetNbinsX(); i++) {
+                bins += " " + std::to_string((int) g->GetBinContent(i));
+              }
+              LOG(INFO) << bins;
             }
-            LOG(INFO) << bins;
           }
         };
       }
diff --git a/Framework/src/runMergerTest.cxx b/Framework/src/runMergerTest.cxx
@@ -61,12 +61,13 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
           histo->Fill(p/(double)producersAmount);
 
           MonitorObject* mo = new MonitorObject("histo", histo, "histo-task");
-          mo->setIsOwner(false);
+          mo->setIsOwner(true);
 
-          processingContext.outputs().snapshot<MonitorObject>(Output{"TST", "HISTO", p + 1}, *mo);
-//        processingContext.outputs().snapshot<MonitorObject>(OutputRef{ "mo", p + 1 }, *mo); // dpl template bug?
-          delete mo;
-          delete histo;
+          TObjArray* array = new TObjArray;
+          array->SetOwner(true);
+          array->Add(mo);
+
+          processingContext.outputs().adopt(Output{"TST", "HISTO", p + 1}, array);
         }
       }
     };
@@ -87,15 +88,15 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
   DataProcessorSpec printer{
     "printer",
     Inputs{
-      { "mo", "TST", "HISTO", 0 }
+      { "moarray", "TST", "HISTO", 0 }
     },
     Outputs{},
     AlgorithmSpec{
       (AlgorithmSpec::InitCallback) [](InitContext&) {
-
         return (AlgorithmSpec::ProcessCallback) [](ProcessingContext& processingContext) mutable {
           LOG(INFO) << "printer invoked";
-          auto mo = processingContext.inputs().get<MonitorObject*>("mo").get();
+          auto moArray = processingContext.inputs().get<TObjArray*>("moarray");
+          auto mo = dynamic_cast<MonitorObject*>(moArray->First());
 
           if (mo->getName() == "histo") {
             auto* g = dynamic_cast<TH1F*>(mo->getObject());

Original file line number	Diff line number	Diff line change
`@@ -31,22 +31,19 @@ namespace core`
`31`	`31`	`ObjectsManager::ObjectsManager(TaskConfig& taskConfig) : mTaskName(taskConfig.taskName)`
`32`	`32`	`{`
`33`	`33`	`startPublishing(&mObjectsList, MonitorObject::SYSTEM_OBJECT_PUBLICATION_LIST);`
	`34`	`+ mMonitorObjects.SetOwner(true);`
`34`	`35`	`}`
`35`	`36`
`36`	`37`	`ObjectsManager::~ObjectsManager()`
`37`	`38`	`{`
`38`		`- for (auto& mMonitorObject : mMonitorObjects) {`
`39`		`- delete mMonitorObject.second;`
`40`		`- }`
`41`		`- mMonitorObjects.clear();`
`42`	`39`	`}`
`43`	`40`
`44`	`41`	`void ObjectsManager::startPublishing(TObject* object, std::string objectName)`
`45`	`42`	`{`
`46`	`43`	`std::string nonEmptyName = objectName.empty() ? object->GetName() : objectName;`
`47`	`44`	`auto* newObject = new MonitorObject(nonEmptyName, object, mTaskName);`
`48`	`45`	`newObject->setIsOwner(false);`
`49`		`- mMonitorObjects[nonEmptyName] = newObject;`
	`46`	`+ mMonitorObjects.Add(newObject);`
`50`	`47`
`51`	`48`	`// update index`
`52`	`49`	`if (objectName != MonitorObject::SYSTEM_OBJECT_PUBLICATION_LIST) {`
`@@ -83,8 +80,10 @@ void ObjectsManager::addCheck(const std::string& objectName, const std::string&`
`83`	`80`
`84`	`81`	`MonitorObject* ObjectsManager::getMonitorObject(std::string objectName)`
`85`	`82`	`{`
`86`		`- if (mMonitorObjects.count(objectName) > 0) {`
`87`		`- return mMonitorObjects[objectName];`
	`83`	`+ TObject* mo = mMonitorObjects.FindObject(objectName.c_str());`
	`84`	`+`
	`85`	`+ if (mo) {`
	`86`	`+ return dynamic_cast<MonitorObject*>(mo);`
`88`	`87`	`} else {`
`89`	`88`	`BOOST_THROW_EXCEPTION(ObjectNotFoundError() << errinfo_object_name(objectName));`
`90`	`89`	`}`