MQ/example9

- FairPrimaryGenerator - added setter of the event number; - FairRunSim - check if run id set before generating it; - FairMQSimDevice - added fSimDeviceId member; based on this different devices will generate events with different event numbers, only one device will send parameters to the parameter server; the fSimDeviceId together with runId are obtained from the parameter server; - ParameterMQServer - is able to generate run id, it is send to back to simulation devices together with the devices counter;
FairRootGroup · May 3, 2018 · 6856315 · 6856315
1 parent 137a5c3
commit 6856315
Show file tree

Hide file tree

Showing 6 changed files with 121 additions and 51 deletions.
diff --git a/base/sim/FairPrimaryGenerator.h b/base/sim/FairPrimaryGenerator.h
@@ -64,6 +64,8 @@ class FairPrimaryGenerator : public TNamed {
     fGenList->Add(generator);
   }
 
+  void SetEventNr(Int_t evtNr) { fEventNr = evtNr; }
+
   /** Public method GenerateEvent
       To be called at the beginning of each event from FairMCApplication.
       Generates an event vertex and calls the ReadEvent methods from the

diff --git a/base/steer/FairRunSim.cxx b/base/steer/FairRunSim.cxx
@@ -166,9 +166,11 @@ void FairRunSim::Init()
   fApp->SetGenerator(fGen);
 
   // Add a Generated run ID to the FairRunTimeDb
-  FairRunIdGenerator genid;
-  // FairRuntimeDb *rtdb= GetRuntimeDb();
-  fRunId = genid.generateId();
+  if ( fRunId == 0 ) {
+      FairRunIdGenerator genid;
+      // FairRuntimeDb *rtdb= GetRuntimeDb();
+      fRunId = genid.generateId();
+  }
   fRtdb->addRun(fRunId);
 
   fFileHeader->SetRunId(fRunId);

diff --git a/examples/MQ/9-PixelDetector/src/devices/FairMQSimDevice.cxx b/examples/MQ/9-PixelDetector/src/devices/FairMQSimDevice.cxx
@@ -42,8 +42,20 @@
 
 using namespace std;
 
+// special class to expose protected TMessage constructor
+class SimMQTMessage : public TMessage
+{
+  public:
+  SimMQTMessage(void* buf, Int_t len)
+    : TMessage(buf, len)
+  {
+    ResetBit(kIsOwner);
+  }
+};
+
 FairMQSimDevice::FairMQSimDevice()
   : FairMQDevice()
+  , fSimDeviceId(0)
   , fUpdateChannelName("updateChannel")
   , fRunSim(NULL)
   , fNofEvents(1)
@@ -97,8 +109,38 @@ void FairMQSimDevice::InitTask()
   }
   // ------------------------------------------------------------------------
 
+  // -----   Negotiate the run number   -------------------------------------
+  // -----      via the fUpdateChannelName   --------------------------------
+  // -----      ask the fParamMQServer   ------------------------------------
+  // -----      receive the run number and sampler id   ---------------------
+  std::string* askForRunNumber = new string("ReportSimDevice");
+  FairMQMessagePtr req(NewMessage(const_cast<char*>(askForRunNumber->c_str()),
+                                  askForRunNumber->length(),
+                                  [](void* /*data*/, void* object) { delete static_cast<string*>(object); },
+                                  askForRunNumber));
+  std::unique_ptr<FairMQMessage> rep(NewMessage());
+
+  unsigned int runId = 0;
+  if (Send(req, fUpdateChannelName) > 0)
+      {
+          if (Receive(rep, fUpdateChannelName) > 0)
+              {
+                  std::string repString = string(static_cast<char*>(rep->GetData()), rep->GetSize());
+                  LOG(INFO) << " -> " << repString.data();
+                  runId = stoi(repString);
+                  repString = repString.substr(repString.find_first_of('_')+1,repString.length());
+                  fSimDeviceId = stoi(repString);
+                  LOG(INFO) << "runId = " << runId << "  ///  fSimDeviceId = " << fSimDeviceId;
+              }
+      }
+  // ------------------------------------------------------------------------
+
+  // ------------------------------------------------------------------------
   if ( fPrimaryGenerator )
-    fRunSim->SetGenerator(fPrimaryGenerator);
+      {
+          fPrimaryGenerator->SetEventNr(fSimDeviceId*fNofEvents); // run n simulations with same run id - offset the event number
+          fRunSim->SetGenerator(fPrimaryGenerator);
+      }
   // ------------------------------------------------------------------------
 
   fRunSim->SetStoreTraj(fStoreTrajFlag);
@@ -110,8 +152,9 @@ void FairMQSimDevice::InitTask()
     }
   }
   // ------------------------------------------------------------------------
-  
+
   // -----   Initialize simulation run   ------------------------------------
+  fRunSim->SetRunId(runId); // run n simulations with same run id - offset the event number
   fRunSim->Init();
   // ------------------------------------------------------------------------
 
@@ -123,9 +166,10 @@ void FairMQSimDevice::PreRun()
 
 bool FairMQSimDevice::ConditionalRun()
 {
-  UpdateParameterServer();
-  fRunSim->Run(fNofEvents);
-  return false;
+    if ( fSimDeviceId == 0 )
+        UpdateParameterServer();
+    fRunSim->Run(fNofEvents);
+    return false;
 }
 
 void FairMQSimDevice::UpdateParameterServer()
@@ -163,7 +207,7 @@ void FairMQSimDevice::SendObject(TObject* obj, std::string chan) {
       if (Receive(rep, chan) > 0)
         {
           std::string repString = string(static_cast<char*>(rep->GetData()), rep->GetSize());
-          printf (" -> %s\n",repString.data());
+          LOG(INFO) << " -> " << repString.data();
         }
     }
 }

diff --git a/examples/MQ/9-PixelDetector/src/devices/FairMQSimDevice.h b/examples/MQ/9-PixelDetector/src/devices/FairMQSimDevice.h
@@ -63,6 +63,7 @@ class FairMQSimDevice : public FairMQDevice
     virtual bool ConditionalRun();
 
  private:
+    UInt_t fSimDeviceId;
     std::string fUpdateChannelName;
 
     FairRunSim*     fRunSim;

diff --git a/parmq/ParameterMQServer.cxx b/parmq/ParameterMQServer.cxx
@@ -21,6 +21,7 @@
 #include "FairParAsciiFileIo.h"
 #include "FairParRootFileIo.h"
 #include "FairParGenericSet.h"
+#include "FairRunIdGenerator.h"
 
 #include "ParameterMQServer.h"
 #include "FairMQLogger.h"
@@ -41,6 +42,8 @@ class ParMQTMessage : public TMessage
 
 ParameterMQServer::ParameterMQServer() :
     fRtdb(FairRuntimeDb::instance()),
+    fRunId(0),
+    fNofSimDevices(0),
     fFirstInputName("first_input.root"),
     fFirstInputType("ROOT"),
     fSecondInputName(""),
@@ -195,56 +198,72 @@ bool ParameterMQServer::ProcessRequest(FairMQMessagePtr& req, int /*index*/)
 
 bool ParameterMQServer::ProcessUpdate(FairMQMessagePtr& update, int /*index*/)
 {
-  gGeoManager = NULL; // FairGeoParSet update deletes previous geometry because of resetting gGeoManager, so let's NULL it
+    gGeoManager = NULL; // FairGeoParSet update deletes previous geometry because of resetting gGeoManager, so let's NULL it
 
-  LOG(DEBUG) << "got process update message!";
-  ParMQTMessage tm(update->GetData(), update->GetSize());
+    std::string* text;
 
-  std::string* text;
+    LOG(DEBUG) << "got process update message with size = " << update->GetSize() << " !";
+    if ( update->GetSize() < 20 )
+        {
+            std::string repString = string(static_cast<char*>(update->GetData()), update->GetSize());
+            LOG(INFO) << "Received string " << repString << " !";
+            if ( fNofSimDevices == 0 ) {
+                FairRunIdGenerator genid;
+                fRunId = genid.generateId();
+            }
+            string messageToSend = to_string(fRunId) + "_" + to_string(fNofSimDevices);
+            text = new string(messageToSend);
+            fNofSimDevices += 1;
+            LOG(INFO) << "Replying with \"" << messageToSend << "\"";
+        }
+    else
+        {
+            ParMQTMessage tm(update->GetData(), update->GetSize());
 
-  // get the run id coded in the description of FairParSet
-  FairParGenericSet* newPar = (FairParGenericSet*)tm.ReadObject(tm.GetClass());
-  std::string parDescr = std::string(newPar->getDescription());
-  uint runId = 0;
-  if ( parDescr.find("RUNID") != std::string::npos )
-    {
-      parDescr.erase(0,parDescr.find("RUNID")+5);
-      runId = atoi(parDescr.data());
-      if ( parDescr.find("RUNID") != std::string::npos )
-        parDescr.erase(0,parDescr.find("RUNID")+5);
-    }
-  fRtdb->initContainers(runId);
+            // get the run id coded in the description of FairParSet
+            FairParGenericSet* newPar = (FairParGenericSet*)tm.ReadObject(tm.GetClass());
+            std::string parDescr = std::string(newPar->getDescription());
+            uint runId = 0;
+            if ( parDescr.find("RUNID") != std::string::npos )
+                {
+                    parDescr.erase(0,parDescr.find("RUNID")+5);
+                    runId = atoi(parDescr.data());
+                    if ( parDescr.find("RUNID") != std::string::npos )
+                        parDescr.erase(0,parDescr.find("RUNID")+5);
+                }
+            fRtdb->initContainers(runId);
 
-  newPar->setChanged(true); // trigger writing to file
-  newPar->setStatic(true); // to get rid of error
-  newPar->Print();
+            newPar->setChanged(true); // trigger writing to file
+            newPar->setStatic(true); // to get rid of error
+            newPar->Print();
 
-  if ( fRtdb->addContainer(newPar) )
-    {
-      text = new string("SUCCESS");
-    }
-  else
-    {
-      text = new string("FAIL");
-    }
+            if ( fRtdb->addContainer(newPar) )
+                {
+                    text = new string("SUCCESS");
+                }
+            else
+                {
+                    text = new string("FAIL");
+                }
 
-  Bool_t kParameterMerged = kTRUE;
-  FairParRootFileIo* parOut = new FairParRootFileIo(kParameterMerged);
-  parOut->open(fOutputName.data());
-  fRtdb->setOutput(parOut);
-  fRtdb->saveOutput();
-  fRtdb->closeOutput();
+            Bool_t kParameterMerged = kTRUE;
+            FairParRootFileIo* parOut = new FairParRootFileIo(kParameterMerged);
+            parOut->open(fOutputName.data());
+            fRtdb->setOutput(parOut);
+            fRtdb->saveOutput();
+            fRtdb->closeOutput();
+        }
 
-  FairMQMessagePtr msg(NewMessage(const_cast<char*>(text->c_str()),
-                                  text->length(),
-                                  [](void* /*data*/, void* object) { delete static_cast<string*>(object); },
-                                  text));
+    FairMQMessagePtr msg(NewMessage(const_cast<char*>(text->c_str()),
+                                    text->length(),
+                                    [](void* /*data*/, void* object) { delete static_cast<string*>(object); },
+                                    text));
 
-  if (Send(msg, fUpdateChannelName) < 0)
-    {
-      return false;
-    }
-  return true;
+    if (Send(msg, fUpdateChannelName) < 0)
+        {
+            return false;
+        }
+    return true;
 }
 
 ParameterMQServer::~ParameterMQServer()

diff --git a/parmq/ParameterMQServer.h b/parmq/ParameterMQServer.h
@@ -58,6 +58,8 @@ class ParameterMQServer : public FairMQDevice
 
   private:
     FairRuntimeDb* fRtdb;
+    int            fRunId;
+    int            fNofSimDevices;
 
     std::string fFirstInputName;
     std::string fFirstInputType;