PathUtils Class Reference

Helper utils for path arithmetics needed in the pEventProcessor. More...

#include <PathUtils.h>

Static Public Member Functions
static std::tuple< PathPtr, PathPtr, PathPtr >	splitPath (const PathPtr &path)
	Split the given part into the input, main and output path (in this order) by looking onto the parallel certificate of the modules.
static ModulePtrList	preparePaths (PathPtr &inputPath, PathPtr &mainPath, PathPtr &outputPath)
	Adds internal zmq modules to the paths.
static ModulePtr	getHistogramManager (PathPtr &inputPath)
	Find the histogram manager in the paths and return it.
static ModulePtrList	getTerminateGloballyModules (const ModulePtrList &modules)
	Return only modules which have the TerminateGlobally Module flag set.
static void	prependModulesIfNotPresent (ModulePtrList *modules, const ModulePtrList &prependModules)
	Prepend given 'prependModules' to 'modules', if they're not already present.
static void	appendModule (PathPtr &path, const ModulePtr &module)
	Prepend given modules to the path.
static void	prependModule (PathPtr &path, const ModulePtr &module)
	Append given modules to the path.

Detailed Description

Helper utils for path arithmetics needed in the pEventProcessor.

Definition at line 19 of file PathUtils.h.

Member Function Documentation

appendModule()

void appendModule ( PathPtr &  path, const ModulePtr &  module  )

static

Prepend given modules to the path.

Definition at line 216 of file PathUtils.cc.

{
  path->addModule(module);
}

getHistogramManager()

ModulePtr getHistogramManager ( PathPtr &  inputPath )

static

Find the histogram manager in the paths and return it.

Definition at line 98 of file PathUtils.cc.

{
  ModulePtr histoManagerModule;
 
 
  for (const ModulePtr& module : inputPath->getModules()) {
    if (module->hasProperties(Module::c_HistogramManager)) {
      // Initialize histogram manager if found in the path
      histoManagerModule = module;
 
    }
  }
 
  return histoManagerModule;
}

getTerminateGloballyModules()

ModulePtrList getTerminateGloballyModules ( const ModulePtrList &  modules )

static

Return only modules which have the TerminateGlobally Module flag set.

Definition at line 197 of file PathUtils.cc.

{
  ModulePtrList tmpModuleList;
  for (const ModulePtr& m : modules) {
    if (m->hasProperties(Module::c_TerminateInAllProcesses))
      tmpModuleList.push_back(m);
  }
  return tmpModuleList;
}

preparePaths()

ModulePtrList preparePaths ( PathPtr &  inputPath, PathPtr &  mainPath, PathPtr &  outputPath  )

static

Adds internal zmq modules to the paths.

Definition at line 113 of file PathUtils.cc.

{
  B2ASSERT("The main part is empty. This is a bug in the framework.",
           mainPath and not mainPath->isEmpty());
 
  ModuleManager& moduleManager = ModuleManager::Instance();
 
  const auto& environment = Environment::Instance();
 
  const auto& socketAddress = environment.getZMQSocketAddress();
  const auto inputSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_input));
  const auto outputSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_output));
  const auto pubSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_pub));
  const auto subSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_sub));
 
  unsigned int maximalWaitingTime = environment.getZMQMaximalWaitingTime();
  unsigned int eventBufferSize = environment.getZMQEventBufferSize();
  unsigned int workerTimeout = environment.getZMQWorkerTimeout();
  bool useEventBackup = environment.getZMQUseEventBackup();
 
  if (inputPath) {
    // Add TXInput after input path
    ModulePtr zmqTxInputModule = moduleManager.registerModule("ZMQTxInput");
    zmqTxInputModule->getParam<std::string>("socketName").setValue(inputSocketAddress);
    zmqTxInputModule->getParam<std::string>("xpubProxySocketName").setValue(pubSocketAddress);
    zmqTxInputModule->getParam<std::string>("xsubProxySocketName").setValue(subSocketAddress);
    zmqTxInputModule->getParam<unsigned int>("workerProcessTimeout").setValue(workerTimeout);
    zmqTxInputModule->getParam<bool>("useEventBackup").setValue(useEventBackup);
    zmqTxInputModule->getParam<unsigned int>("maximalWaitingTime").setValue(maximalWaitingTime);
    appendModule(inputPath, zmqTxInputModule);
 
    // Add RXWorker before main path
    ModulePtr zmqRxWorkerModule = moduleManager.registerModule("ZMQRxWorker");
    zmqRxWorkerModule->getParam<std::string>("socketName").setValue(inputSocketAddress);
    zmqRxWorkerModule->getParam<std::string>("xpubProxySocketName").setValue(pubSocketAddress);
    zmqRxWorkerModule->getParam<std::string>("xsubProxySocketName").setValue(subSocketAddress);
    zmqRxWorkerModule->getParam<unsigned int>("maximalWaitingTime").setValue(maximalWaitingTime);
    zmqRxWorkerModule->getParam<unsigned int>("eventBufferSize").setValue(eventBufferSize);
    prependModule(mainPath, zmqRxWorkerModule);
 
  }
 
  if (outputPath) {
    // Add TXWorker after main path
    ModulePtr zmqTxWorkerModule = moduleManager.registerModule("ZMQTxWorker");
    zmqTxWorkerModule->getParam<std::string>("socketName").setValue(outputSocketAddress);
    zmqTxWorkerModule->getParam<std::string>("xpubProxySocketName").setValue(pubSocketAddress);
    zmqTxWorkerModule->getParam<std::string>("xsubProxySocketName").setValue(pubSocketAddress);
    appendModule(mainPath, zmqTxWorkerModule);
 
    // Add RXOutput before output path
    ModulePtr zmqRxOutputModule = moduleManager.registerModule("ZMQRxOutput");
    zmqRxOutputModule->getParam<std::string>("socketName").setValue(outputSocketAddress);
    zmqRxOutputModule->getParam<std::string>("xpubProxySocketName").setValue(pubSocketAddress);
    zmqRxOutputModule->getParam<std::string>("xsubProxySocketName").setValue(subSocketAddress);
    zmqRxOutputModule->getParam<unsigned int>("maximalWaitingTime").setValue(maximalWaitingTime);
    prependModule(outputPath, zmqRxOutputModule);
 
  }
 
  if (inputPath) {
    B2INFO("Input Path " << inputPath->getPathString());
  }
  if (mainPath) {
    B2INFO("Main Path " << mainPath->getPathString());
  }
  if (outputPath) {
    B2INFO("Output Path " << outputPath->getPathString());
  }
 
  Path mergedPath;
  if (inputPath) {
    mergedPath.addPath(inputPath);
  }
  mergedPath.addPath(mainPath);
  if (outputPath) {
    mergedPath.addPath(outputPath);
  }
 
  B2INFO("ModuleList " << mergedPath.getPathString());
 
  return mergedPath.buildModulePathList();
}

prependModule()

void prependModule ( PathPtr &  path, const ModulePtr &  module  )

static

Append given modules to the path.

Definition at line 221 of file PathUtils.cc.

{
  PathPtr newPath(new Path());
  newPath->addModule(module);
  newPath->addPath(path);
  path.swap(newPath);
}

prependModulesIfNotPresent()

void prependModulesIfNotPresent ( ModulePtrList *  modules, const ModulePtrList &  prependModules  )

static

Prepend given 'prependModules' to 'modules', if they're not already present.

Definition at line 207 of file PathUtils.cc.

{
  for (const ModulePtr& m : prependModules) {
    if (std::find(modules->begin(), modules->end(), m) == modules->end()) { //not present
      modules->push_front(m);
    }
  }
}

splitPath()

std::tuple< PathPtr, PathPtr, PathPtr > splitPath ( const PathPtr &  path )

static

Split the given part into the input, main and output path (in this order) by looking onto the parallel certificate of the modules.

Definition at line 16 of file PathUtils.cc.

{
  //modules that can be parallelized, but should not go into a parallel section by themselves
  std::set<std::string> uselessParallelModules({"HistoManager", "Gearbox", "Geometry"});
 
  PathPtr inputPath(new Path);
  PathPtr mainPath(new Path);
  PathPtr outputPath(new Path);
 
  int stage = 0; //0: in, 1: event/main, 2: out
  for (const ModulePtr& module : path->getModules()) {
    bool hasParallelFlag = module->hasProperties(Module::c_ParallelProcessingCertified) and
                           !module->hasProperties(Module::c_Input) and
                           !module->hasProperties(Module::c_Output) and
                           !module->hasProperties(Module::c_HistogramManager) ;
    //entire conditional path must also be compatible
    if (hasParallelFlag and module->hasCondition()) {
      for (const auto& conditionPath : module->getAllConditionPaths()) {
        if (!ModuleManager::allModulesHaveFlag(conditionPath->getModules(), Module::c_ParallelProcessingCertified)) {
          hasParallelFlag = false;
        }
      }
    }
    //if modules have parallal flag -> stage = 1 , event/main
    if ((stage == 0 and hasParallelFlag) or (stage == 1 and !hasParallelFlag)) {
      stage++;
 
      if (stage == 2) {
        bool path_is_useful = false;
        for (const auto& parallelModule : mainPath->getModules()) {
          if (uselessParallelModules.count(parallelModule->getType()) == 0) {
            path_is_useful = true;
            break;
          }
        }
        if (not path_is_useful) {
          //merge mainPath back into input path
          inputPath->addPath(mainPath);
          mainPath.reset(new Path);
          //and search for further parallel sections
          stage = 0;
        }
      }
    }
 
    if (stage == 0) {
      inputPath->addModule(module);
      if (module->hasProperties(Module::c_HistogramManager)) {
        // Initialize histogram manager if found in the path
 
        //add histoman to other paths
        mainPath->addModule(module);
        outputPath->addModule(module);
      }
    } else if (stage == 1) {
      mainPath->addModule(module);
    } else if (stage == 2) {
      outputPath->addModule(module);
    }
  }
 
  bool createAllPaths = false; //usually we might not need e.g. an output path
  for (const ModulePtr& module : path->getModules()) {
    if (module->hasProperties(Module::c_TerminateInAllProcesses)) {
      createAllPaths = true; //ensure there are all kinds of processes
    }
  }
 
  // if main path is empty, createAllPaths doesn't really matter, since we'll fall back to single-core processing
  if (mainPath->isEmpty() and not createAllPaths) {
    mainPath.reset();
  }
  if (inputPath->isEmpty() and not createAllPaths) {
    inputPath.reset();
  }
  if (outputPath->isEmpty() and not createAllPaths) {
    outputPath.reset();
  }
 
  return {inputPath, mainPath, outputPath};
}

---

The documentation for this class was generated from the following files:

• framework/pcore/include/PathUtils.h
• framework/pcore/src/PathUtils.cc