JobCPath.cxx

Go to the documentation of this file.

// $Id: JobCPath.cxx,v 1.65 2007/10/25 17:21:25 bckhouse Exp $
//
// A collection of modules linked together in the order they are to be
// executed
//
// messier@huhepl.harvard.edu
#include "JobControl/JobCPath.h"
#include <cstring>  // strcmp
#include <vector>
#ifdef SITE_HAS_SIGC
# include "sigc++/class_slot.h"
#endif
#include "MessageService/MsgService.h"
#include "MessageService/MsgFormat.h"
#include "MessageService/MsgTripWire.h"
#include "JobControl/JobCEnv.h"
#include "JobControl/JobCModuleRegistry.h"
#include "JobControl/JobCModuleProxy.h"
#include "JobControl/JobCRecord.h"
#include "JobControl/JobCModule.h"
#include "JobControl/JobCMethod.h"
#include "JobControl/JobCommand.h"
#include "JobControl/JobCInput.h"

using namespace std;

CVSID("$Id: JobCPath.cxx,v 1.65 2007/10/25 17:21:25 bckhouse Exp $");

ClassImp(JobCPath)

class JobCPathConnectionHelper
{
//======================================================================
// A little helper class to make a list of paths connected to this one
//======================================================================
public:
  JobCPathConnectionHelper(const JobCPath* p) {
    this->BuildConnectionList(p);
    this->PreenConnectionList();
  }
  void BuildConnectionList(const JobCPath* p1) {
    // Recursively find all paths attached to p1
    fConnectionList.push_back(p1);
    for (int i=0; void *v=p1->GetAttached(i); ++i) {
      JobCPath *p2 = static_cast<JobCPath*>(v);
      this->BuildConnectionList(p2);
    }
  }
  void PreenConnectionList() {
    // Eliminate duplicates, keeping the deeper of the two occurances
    int size = fConnectionList.size();
    for (int i=0; i<size-1; ++i) {
      for (int j=i+1; j<size; ++j) {
        if (fConnectionList[i] == fConnectionList[j]) fConnectionList[i] = 0;
      }
    }
  }
  ostream& DoPrint(ostream& os) {
    // Print everyone in the list
    vector<const JobCPath*>::iterator itr(fConnectionList.begin());
    vector<const JobCPath*>::iterator itrEnd(fConnectionList.end());
    for (; itr!=itrEnd; ++itr) if (*itr) (*itr)->Print(os);
    return os;
  }
  void DoHandleResult(const JobCResult& r) {
    // Alert everyone in the list to the status change
    vector<const JobCPath*>::iterator itr(fConnectionList.begin());
    vector<const JobCPath*>::iterator itrEnd(fConnectionList.end());
    for (; itr!=itrEnd; ++itr) {
      if (*itr) (*itr)->HandleResult(r);
    }
  }
  void DoBeginJob() {
    // Alert everyone in the list to the start of job
    vector<const JobCPath*>::iterator itr(fConnectionList.begin());
    vector<const JobCPath*>::iterator itrEnd(fConnectionList.end());
    for (; itr!=itrEnd; ++itr) {
      if (*itr) (*itr)->BeginJob();
    }
  }
  void SetRefreshMode(bool m) {
    // Alert everyone in the list that we're running in refresh mode
    vector<const JobCPath*>::iterator itr(fConnectionList.begin());
    vector<const JobCPath*>::iterator itrEnd(fConnectionList.end());
    for (; itr!=itrEnd; ++itr) {
      if (*itr) (*itr)->SetRefreshMode(m);
    }
  }
private:
  vector<const JobCPath*> fConnectionList; // List of job paths connected
};

//......................................................................

void JobCPath::Print(ostream& os) const
{
//======================================================================
// Print the job path to the MsgStream os
//======================================================================
  os << fName.c_str()
     << "("
     << fNin   << " in "
     << fNpass << " out "
     << fNfail << " filt.)\n";
  int i=0;
  MsgFormat i1("%3d");
  list<JobCNode*>::const_iterator      itrEnd      = fNodeList.end();
  for (list<JobCNode*>::const_iterator itrJobCNode = fNodeList.begin();
       itrJobCNode != itrEnd;
       ++itrJobCNode) {
    os << i1(++i) << ") " << (**itrJobCNode) << "\n";
  }

  // Print list of paths attached to this one
  bool haveconnect = false;
  const JobCGraphVtx* v = 0;
  for (i=0; (v=this->GetAttached(i))!=0; ++i) {
    const JobCPath* p2 = dynamic_cast<const JobCPath*>(v);
    if (p2!=0 && haveconnect==false) {
      os << "  |-->";
    }
    os << "/" << p2->GetName();
    haveconnect = true;
  }
  if (haveconnect) os << "/\n";
}

//......................................................................

ostream& operator<<(ostream& os, const JobCPath& p)
{
  JobCPathConnectionHelper jcpph(&p);
  return jcpph.DoPrint(os);
}

//......................................................................

JobCPath::JobCPath() :
  fIsRefreshRun(false),
  fAtJobStart(true),
  fName(""),
  fNin(0),
  fNpass(0),
  fNfail(0),
  fMom(0),
  fInput(0),
  fDummyModule(0),
  fDummyNode(0)
{
    MSG("JobC", Msg::kDebug) << " default constructor\n";
    //cerr << "JobCPath::JobCPath()\n";
}

//......................................................................

JobCPath::JobCPath(const char*      name,
                   MomNavigator*    mom,
                   JobCInputModule* inp) :
  fIsRefreshRun(false),
  fAtJobStart(true),
  fName(name),
  fNin(0),
  fNpass(0),
  fNfail(0),
  fMom(mom),
  fInput(inp),
  fDummyModule(0),
  fDummyNode(0)
{
//======================================================================
// Purpose: Create and empty, named path which lives in the job
//          environment "env"
// Input  : name - the name of this path
//        : env  - pointer to the job environment data
//======================================================================

    MSG("JobC", Msg::kDebug) << " input module at " << (void*)inp << endl;
    //cerr << "JobCPath::JobCPath(\""<<name<<"\","<<(void*)mom<<","<<(void*)inp<<")\n";
}

//......................................................................

JobCPath::~JobCPath()
{
//======================================================================
// Purpose: Delete a path
//======================================================================
  MSG("JobC", Msg::kVerbose) << this->GetName() << " destructor" << "\n";

#ifdef SITE_HAS_SIGC
  this->SigDelete(this);
#endif

  // Delete the nodes attached to this path
  list<JobCNode*>::iterator itrJobCNode;
  for (itrJobCNode = fNodeList.begin();
       itrJobCNode != fNodeList.end();
       ++itrJobCNode) {
    MSG("JobC", Msg::kVerbose) <<
      " >delete node " << (**itrJobCNode) << "\n";
    delete (*itrJobCNode);
  }

  // Delete modules attached to this path
  list<JobCModule*>::iterator itrJobCModule;
  for (itrJobCModule  = fModuleList.begin();
       itrJobCModule != fModuleList.end();
       ++itrJobCModule) {
    MSG("JobC", Msg::kVerbose) <<
      " >delete module " << (*itrJobCModule)->GetName() << "\n";

    // Tell the module the job is over before deleting it
    (*itrJobCModule)->EndJob();

    delete (*itrJobCModule);
  }

  if (fDummyModule) { delete fDummyModule; fDummyModule = 0; }
  if (fDummyNode)   { delete fDummyNode;   fDummyNode   = 0; }
}

//......................................................................

#ifdef SITE_HAS_SIGC
void JobCPath::NodeUpdateCB(JobCNode* /*n*/, JobCNode::Update_t u)
{
//======================================================================
// Provide a call-back to handle node reconfiguration
//======================================================================
  switch (u) {
  case JobCNode::kFilter: this->SigUpdate(this,JobCPath::kFilters);
  default:                this->SigUpdate(this,JobCPath::kGeneral);
  }
}
#endif

//......................................................................

bool JobCPath::AtJobStart() const
{
//======================================================================
// Are we starting a new job?
//======================================================================
  bool tmp = fAtJobStart;
  fAtJobStart = false;
  return tmp;
}

//......................................................................

void JobCPath::CheckResult(const JobCResult& r)
{
//======================================================================
// Purpose: Check the result of a node to see if the path needs to
//          emit any status changes
//======================================================================
  // Check conditions that require a response
    MSG("JobC", Msg::kVerbose) << "CheckResult: Current result is " << r << endl;

  if ( r.BeginFile() || r.EndFile() || r.BeginRun()  || r.EndRun() ) {
    this->PropagateResult(r,this);
  }
}

//......................................................................

void JobCPath::HandleResult(const JobCResult& r) const
{
//======================================================================
// Purpose: Check the result of a node to see if the path needs to
//          emit any status changes
//======================================================================
    MSG("JobC", Msg::kVerbose) << "HandleResult: Current result is " << r << endl;

  bool beginFile = r.BeginFile();
  bool endFile   = r.EndFile();
  bool beginRun  = r.BeginRun();
  bool endRun    = r.EndRun();

  list<JobCModule*>::const_iterator itrJobCModule;
  for (itrJobCModule  = fModuleList.begin();
       itrJobCModule != fModuleList.end();
       ++itrJobCModule) {
    if (endFile)   { (*itrJobCModule)->EndFile();   }
    if (endRun)    { (*itrJobCModule)->EndRun();    }
    if (beginRun)  { (*itrJobCModule)->BeginRun();  }
    if (beginFile) { (*itrJobCModule)->BeginFile(); }
  }
}

//......................................................................

void JobCPath::PropagateResult(const JobCResult& r, JobCPath* p)
{
//======================================================================
// Send the result r to all the paths attached to path p
//======================================================================
  // Use the print helper to get the list of attached paths
  JobCPathConnectionHelper h(p);
  h.DoHandleResult(r);
}

//......................................................................

const char* JobCPath::GetName() const { return fName.c_str(); }

//......................................................................

JobCModule *JobCPath::GetModule(const char *moduleName, bool create)
{
//======================================================================
// Get the module named "moduleName" in this path. If no module is
// found then the method returns 0 if create==false. If create==true
// the a module with the given name is created and a pointer to it
// returned.
//======================================================================
  list<JobCModule*>::iterator itrJobCModule(fModuleList.begin());
  list<JobCModule*>::iterator itrEnd(fModuleList.end());
  for (;itrJobCModule != itrEnd; ++itrJobCModule) {
    if (strcmp(moduleName,(*itrJobCModule)->GetName())==0) {
      MSG("JobC",Msg::kVerbose) <<
        "Found " << moduleName << " in path " << fName << "\n";
      return (*itrJobCModule);
    }
  }
  // Failed to find the module. Either return 0 or build the module
  if (create == false) return 0;

  // Not found and we've been asked to make one.
  JobCModuleProxy*
    jmp = JobCModuleRegistry::Instance().LookUp(moduleName);
  if (jmp) {
    JobCModule* m = jmp->CreateModule(); // Remember to delete this!
    m->SetPath(this);
    fModuleList.push_back(m);            // Add to path's modules
    return m;
  }

  // Errors fall through to here
  return 0;
}

//......................................................................

JobCNode* JobCPath::FindNode(const char* module, const char* method)
{
//======================================================================
// Purpose: Look for a node which matches the module method pair
//======================================================================
  list<JobCNode*>::const_iterator itr(fNodeList.begin());
  list<JobCNode*>::const_iterator itrEnd(fNodeList.end());
  for (; itr!=itrEnd; ++itr) {
    if ((*itr)->MatchModuleMethod(module,method)) {
      return (*itr);
    }
  }
  return 0;
}

JobCInputModule* JobCPath::GetInputModule()
{
    return fInput;
}


//......................................................................

bool JobCPath::HandleError(const JobCResult& result, const JobCNode* n)
{
//======================================================================
// Purpose: Handle any errors that might have occured from executing a
// node
//
// Returns: true  - stop the run
//          false - continue the run
//======================================================================
  JobCResult::Error_t errlvl = result.ErrorLevel();
  if (errlvl == JobCResult::kWarning) {
//    MSG("JobC",Msg::kWarning) << "Warning in: [" << (*n) << "]\n";
  }
  else if (errlvl == JobCResult::kError) {
    MSG("JobC",Msg::kError) << "Error in: [" << (*n) << "]\n";
  }
  else {
    MSG("JobC",Msg::kFatal) <<
      "Fatal error in: [" << (*n) << "]\n";
  }

  // Is error relatively mild?
  JobCResult::Error_t errorThreshold = JobCResult::kError;
  if (errlvl < errorThreshold) return false; // Keep rolling...

  // Is the erro fatal?
  if (errlvl == JobCResult::kFatal) abort();

  return true; // Stop the run
}

//......................................................................

JobCModule& JobCPath::Mod(const char* moduleName)
{
//======================================================================
// Return the module in this path by name. If not found return a dummy
// module. This method is intened for use interactively so it doesn't
// use pointers. Ie. it tries to help prevent people from doing
// (JobCModule*)0->Method(); which would trash their root session
//======================================================================
  JobCModule* m = this->GetModule(moduleName, false);
  if (m) { return (*m); }

  MSG("JobC",Msg::kWarning) <<
    "Path '" << this->GetName() << "' " <<
    "does not contain module '" << moduleName << "'.\n";

  // In this case return a dummy, empty module to avoid having things
  // complete blow up in the user's face...
  if (fDummyModule == 0) fDummyModule = new JobCModule;
  return (*fDummyModule);
}

//......................................................................

JobCNode& JobCPath::Node(const char* nodeName)
{
//======================================================================
// Return the module in this path by name. If not found return a dummy
// module. This method is intened for use interactively so it doesn't
// use pointers. Ie. it tries to help prevent people from doing
// (JobCModule*)0->Method(); which would trash their root session
//======================================================================
  std::string module;
  std::string method;
  JobCommand::SplitLine(nodeName,':',module,method);

  JobCNode* n = this->FindNode(module.c_str(),method.c_str());
  if (n) { return (*n); }

  MSG("JobC",Msg::kWarning) <<
    "Path '" << this->GetName() << "' " <<
    "does contain node '" << nodeName << "'.\n";

  // In this case return a dummy, empty module to avoid having things
  // complete blow up in the user's face...
  if (fDummyNode == 0) fDummyNode = new JobCNode;
  return (*fDummyNode);
}

//......................................................................

JobCNode* JobCPath::CreateNode(const char* modName, const char* metName)
{
//======================================================================
// Purpose: Create a node using a specified module and method
//
// Inputs: modName - name of the module to use
//         metName - name of the method to use
//
// Returns: Pointer to the node created
//======================================================================
  // Look the module up
  JobCModule *module = this->GetModule(modName, true);
  if (module == 0) {
    MSG("JobC",Msg::kWarning) <<
      "Get failed for module '" << modName << "'.\n";
    return 0;
  }

  // Get the method
  const JobCMethod* method = JobCMethod::GetMethodByName(metName);
  if (method == 0) {
    MSG("JobC",Msg::kWarning) <<
      "Look up failed for method '" << metName << "'\n";
    return 0;
  }
  MSG("JobC", Msg::kDebug) << "Created node "
                           << module->GetName() << "::"
                           << method->GetName() << "\n";

  // Create the node
  JobCNode* n = new JobCNode(module, method);
#ifdef SITE_HAS_SIGC
  if (n) {
    // Attach the Filter update signal to the node's filter update signal
    n->SigUpdate.connect(SigC::slot_class(*this,&JobCPath::NodeUpdateCB));
  }
#endif
  return n;
}

//......................................................................

void JobCPath::SetAllFilters(bool onOff)
{
//======================================================================
// Purpose: Set all nodes to filter (true) or not to filter (false)
//======================================================================
  list<JobCNode*>::iterator itr(fNodeList.begin());
  list<JobCNode*>::iterator itrEnd(fNodeList.end());
  for (; itr!=itrEnd; ++itr) {
    if (onOff) {
      (*itr)->FilterOn();
    }
    else {
      (*itr)->FilterOff();
    }
  }
}

//......................................................................

void JobCPath::ReverseAllFilters()
{
//======================================================================
// Purpose: Reverse the meaning of pass/fail for all active filters
//======================================================================
  list<JobCNode*>::iterator itr(fNodeList.begin());
  list<JobCNode*>::iterator itrEnd(fNodeList.end());
  for (; itr!=itrEnd; ++itr) {
    (*itr)->ReverseFilter();
  }
}

//......................................................................

JobCNode* JobCPath::PushFront(const char *moduleName, const char *methodName)
{
//======================================================================
// Purpose: Add a module::method pair to the begining of the path
//
// Inputs: moduleName - name of the module
//         methodName - name of the method
//
// Returns: pointer to node created
//======================================================================
  // Create a node
  JobCNode *node = this->CreateNode(moduleName, methodName);

  // Add the node to the front of the list
  if (node) {
    fNodeList.push_front(node);
#ifdef SITE_HAS_SIGC
    this->SigUpdate(this,JobCPath::kAddNode);
#endif
  }
  return node;
}

//......................................................................

JobCNode* JobCPath::PushBack(const char *moduleName, const char *methodName)
{
//======================================================================
// Purpose: Add a module::method pair to the end of the path
//
// Inputs: moduleName - name of the module
//         methodName - name of the method
//
// Returns: pointer to the node created
//======================================================================
  // Create a node
  JobCNode *node = this->CreateNode(moduleName, methodName);
  if (node == 0) {
    MSG("JobC", Msg::kWarning)
      << "Failed to create node "
      << moduleName << "::" << methodName << "\n";
    return 0;
  }

  // Add the node to the front of the list
  if (node) {
    fNodeList.push_back(node);
#ifdef SITE_HAS_SIGC
    this->SigUpdate(this,JobCPath::kAddNode);
#endif
  }
  return node;
}

//......................................................................

JobCNode* JobCPath::AddAt(const char *moduleName,
                          const char *methodName, int n)
{
//======================================================================
// Purpose: Add a module::method pair at the nth position in the path
//
// Inputs: moduleName - name of the module
//         methodName - name of the method
//
// Returns: pointer to node created
//======================================================================
  // Create the node
  JobCNode *node = this->CreateNode(moduleName, methodName);

  // Advance to this place in the list. Note: start at 1 since that's
  // the way humans count. Ie. n=1 places the node first in the list,
  // n=2 second in the list etc.
  list<JobCNode*>::iterator itrJobCNode = fNodeList.begin();
  for (int i=1; i<n; ++i) {
    if (++itrJobCNode == fNodeList.end()) break;
  }

  // Insert at this point
  if (node) {
    fNodeList.insert(itrJobCNode, node);
#ifdef SITE_HAS_SIGC
    this->SigUpdate(this,JobCPath::kAddNode);
#endif
  }

  return node;
}

//......................................................................

JobCNode* JobCPath::AddBetween(const char *moduleName,
                               const char *methodName,
                               const char *afterNodeName,
                               const char *beforeNodeName)
{
//======================================================================
// Purpose: Add a module::method pair to the path between named modules
//
// Inputs: moduleName     - name of the module
//         methodName     - name of the method
//         afterNodeName  - name of the node to insert after
//         beforeNodeName - name of the node to insert before
//
// Returns: pointer to node created
//======================================================================
  // Create the node.
  JobCNode* newNode = this->CreateNode(moduleName, methodName);

  if(!newNode){
    MSG("JobCPath", Msg::kFatal) << "Could not create node "
                                 << moduleName << ":" << methodName << endl;
    return 0;
  }

  // Find the node we have to insert after.
  std::string afterModule, afterMethod;
  JobCommand::SplitLine(afterNodeName, ':', afterModule, afterMethod);
  JobCNode* afterNode = this->FindNode(afterModule.c_str() ,afterMethod.c_str());

  // Find the node we have to insert before.
  std::string beforeModule, beforeMethod;
  JobCommand::SplitLine(beforeNodeName, ':', beforeModule, beforeMethod);
  JobCNode* beforeNode = this->FindNode(beforeModule.c_str() ,beforeMethod.c_str());

  if(!afterNode){
    MSG("JobCPath", Msg::kFatal) << "Could not find node "
                                 << afterNodeName << endl;
    return 0;
  }
  if(!beforeNode){
    MSG("JobCPath", Msg::kFatal) << "Could not find node "
                                 << beforeNodeName << endl;
    return 0;
  }

  list<JobCNode*>::iterator it = fNodeList.begin();

  for(; it != fNodeList.end(); ++it){
    if(*it == afterNode){ // We found the node to go after.
      // Find the node we will be before.
      ++it;
      // Check it's the one we ought to be before.
      if(it != fNodeList.end() && *it == beforeNode){
        fNodeList.insert(it, newNode);
#ifdef SITE_HAS_SIGC
        this->SigUpdate(this, JobCPath::kAddNode);
#endif
        return newNode;
      }
      else{
        MSG("JobCPath", Msg::kFatal) << afterNodeName
                                     << " and "
                                     << beforeNodeName
                                     << " are not adjacent\n";
        return 0;
      }
    }
  }

  MSG("JobCPath", Msg::kFatal) << "This should never happen.\n";
  return 0;
}

//......................................................................

void JobCPath::Remove(const char *moduleName, const char *methodName)
{
//======================================================================
// Purpose: Remove a module::method pair from the path
//
// Inputs: moduleName - name of the module
//         methodName - name of the method
//======================================================================
  // Find the node coresponding to this module/method
  bool haveMatch = false; // Is there a node that matches
  list<JobCNode*>::iterator itrJobCNode(fNodeList.begin());
  list<JobCNode*>::iterator itrEnd(fNodeList.end());
  for (;itrJobCNode != itrEnd; ++itrJobCNode) {
    if ( (haveMatch =
          (*itrJobCNode)->MatchModuleMethod(moduleName, methodName))) {
      break;
    }
  }

  if (haveMatch) {
    // Delete the node object pointed to
    delete (*itrJobCNode);

    // Remove pointer from list
    fNodeList.erase(itrJobCNode);
#ifdef SITE_HAS_SIGC
    this->SigUpdate(this,JobCPath::kRemoveNode);
#endif
    return;
  }

  // No match in list
  MSG("JobC", Msg::kWarning)
    << "Path does not contain "
    << moduleName << "::" << methodName << ".\n";
}

//......................................................................

bool JobCPath::EvalInput(JobCRecord* rec)
{
//======================================================================
// Evaluate this node in the graph
//======================================================================
  // Get the result of running this path
  JobCResult result = this->Execute(rec);
  MSG("JobC", Msg::kVerbose) << "EvalInput gets result = " << result << endl;



  // Decide if execute was sucessful or not
  if ( result.EndOfInputStream() || result.Failed() ) return false;
  return true;
}

//......................................................................

void JobCPath::Attach(JobCGraphVtx *p)
{
//======================================================================
// Attach a path to this one
//======================================================================
  this->JobCGraphVtx::Attach(p);
#ifdef SITE_HAS_SIGC
  this->SigUpdate(this,JobCPath::kAddConnection);
#endif
}

//......................................................................

void JobCPath::Detach(JobCGraphVtx *p)
{
//======================================================================
// Attach a path to this one
//======================================================================
  this->JobCGraphVtx::Detach(p);
#ifdef SITE_HAS_SIGC
  this->SigUpdate(this,JobCPath::kRemoveConnection);
#endif
}

//......................................................................

JobCResult JobCPath::Execute(JobCRecord* rec)
{
//======================================================================
// Purpose: Run a record through every node in the path
//
// Inputs: rec - the MINOS Object Map that holds the data to process
//
// Returns: Pass/Fail depending on whether the record passed every node
//======================================================================

  // Count events processed
  ++fNin;

  // Loop over all nodes in path
  list<JobCNode*>::iterator itrJobCNode(fNodeList.begin());
  list<JobCNode*>::iterator itrEnd(fNodeList.end());
  for (;itrJobCNode != itrEnd; ++itrJobCNode) {

    // We'll need to make some querries about what kind of node this
    // is, so get the method type
    JobCMethod m = (*itrJobCNode)->GetMethod();

    // Don't evalute Get methods on "refresh" runs
    // This makes sense in the "old" framework (hence the fInput=0
    if (fInput==0 && (fIsRefreshRun == true) && (m == JobCMethod::kGet)) {
      continue;
    }

    // Evaluate this node
    MsgTripWire::Instance().StartNode(); //Record start of node
    fResult = (*itrJobCNode)->Execute(rec->GetDataPtr());
    MSG("JobC", Msg::kVerbose) << "Result of node " << *(*itrJobCNode) << "execution is: " << fResult << endl;


    // See if the job module produced any errors, warnings, or other
    // conditions which require special handling
    if (fResult.HaveError()) {
      bool stopRun = this->HandleError(fResult,(*itrJobCNode));
      if (stopRun) return fResult;
    }

    // On end of input we have to adjust the counter since the stream
    // failed to deliver a new record
    if ( fResult.EndOfInputStream() ) {
      --fNin;
      return fResult;
    }

    // See if record was labeled non-phyiscs
    if ( fResult.NonPhysicsRecord() ) rec->SetIsPhysics(false);

    // See if this result has triggered any status changes
    this->CheckResult(fResult);

    // Exit on failure
    if (fResult.Failed()) {
      ++fNfail;
      return fResult;
    }
  } // loop on nodes

  // Passed all nodes
  ++fNpass;
  return fResult;
}

//......................................................................

void JobCPath::BeginJob() const
{
//======================================================================
// Purpose: Signal BeginJob for every node in the path
//
//======================================================================

  // First time entering a path, send the begin job message to each
  // module in the path
  list<JobCModule*>::const_iterator i(fModuleList.begin());
  list<JobCModule*>::const_iterator iEnd(fModuleList.end());
  for (; i!=iEnd; ++i) if (*i) (*i)->BeginJob();

}

//......................................................................

void JobCPath::SetRefreshMode(bool m) const
{
  // This bool is mutable so the const is OK
  fIsRefreshRun = m;
}

//......................................................................

void JobCPath::PropagateRefreshMode(bool m)
{
//======================================================================
// Tell this path and all others attached that we're in refresh mode m
//======================================================================
  JobCPathConnectionHelper h(this);
  h.SetRefreshMode(m);
}

//......................................................................

void JobCPath::Report() { MSG("JobCReport",Msg::kInfo) << (*this) << "\n"; }

//......................................................................

void JobCPath::Run(int n, JobCPath::RunMode_t runmode)
{
//======================================================================
// Purpose: Run records through the path until the path cannot find
//          records to process
//======================================================================
  assert(fMom);
  JobCRecord rec(fMom);

  this->SetUniquePaths();

#ifdef SITE_HAS_SIGC
  this->SigStartRun(this);
#endif

  // n = 0 is a flag to re-run the current event
  if (n == 0) {
    this->PropagateRefreshMode(true);
    this->EvaluateGraph(&rec);
    this->PropagateRefreshMode(false);
#ifdef SITE_HAS_SIGC
    this->SigEndRun(this);
#endif
    return;
  }
  else {

    // First time entering a path, send the begin job message to each
    // module in the path, as well as the attached paths.
    // Use the path helper to process the list of attached paths.
    if (this->AtJobStart()) {
      JobCPathConnectionHelper h(this);
      h.DoBeginJob();
    }

    // n != 0 runs new events
    bool done  = false;  // Are we done with the run?
    int nIn0   = fNin;   // Number events already read at start of run
    int nPass0 = fNpass; // Number events already passed at start of run
    int nFail0 = fNfail; // Number events already failed at start of run
    while (done == false) {
      MsgTripWire::Instance().StartRecordSet(); //Record start of record set
      rec.Reset();
      if (fInput) {
        fResult = fInput->Next();
        // Reset the prefix on future MsgService calls.
        (MsgService::Instance())->SetCurrentRunSnarl(fInput->GetCurrentRun(),
                                                     fInput->GetCurrentSnarl());

        MSG("JobC", Msg::kVerbose) << "Result input module is: "
                                   << fResult << endl;
        this->CheckResult(fResult);

        if (fResult.EndOfInputStream()) {
          done |= true;
          break;
        }
      }
      this->EvaluateGraph(&rec);

      // Decide if we're done based on number of events processed
      // The variables fNin, fNpass, and fNfail get updated as we run
      done |= ( (runmode == JobCPath::kRunAll)   && (false) );
      done |= ( (runmode == JobCPath::kRunNin)   && (fNin  -nIn0  )==n );
      done |= ( (runmode == JobCPath::kRunNpass) && (fNpass-nPass0)==n );
      done |= ( (runmode == JobCPath::kRunNfail) && (fNfail-nFail0)==n );

      // Check if we've reached the end of the input stream
      if (fResult.EndOfInputStream()) done |= true;

      // Check run limits from the job environmemt
      if (JobCEnv::Instance().ContinueRun(fNin) == false) {
        done |= true;
      }

      // Prevent running forever...
      if ( (fNin-nIn0)==JOBCPATH_RUN_LIMIT ) {
        MSG("JobC",Msg::kWarning) <<
          "Run limit (" << JOBCPATH_RUN_LIMIT << " records) reached.\n";
        done |= true;
      }
    }
  }
#ifdef SITE_HAS_SIGC
  this->SigEndRun(this);
#endif
}

//......................................................................

void JobCPath::RunNin(int n)
{
//======================================================================
// Purpose: Run n records through the path - n=0 is a special case for
//          interactive jobs
//======================================================================
  this->Run(n,JobCPath::kRunNin);
}

//......................................................................

void JobCPath::RunNpass(int n)
{
//======================================================================
// Purpose: Run records through the path until n pass
//======================================================================
  this->Run(n,JobCPath::kRunNpass);
}

//......................................................................

void JobCPath::RunNfail(int n)
{
//======================================================================
// Purpose: Run records through the path until n fail
//======================================================================
  this->Run(n,JobCPath::kRunNfail);
}
void JobCPath::SetUniquePaths(const char* base)
{
    string path = Form("%s/%s",base,fName.c_str());

    list<JobCNode*>::const_iterator nit, done =  fNodeList.end();
    for (nit = fNodeList.begin(); nit != done; ++nit) {
        JobCModule* mod = (*nit)->GetModule();
        string name = Form("%s/%s",path.c_str(),mod->GetName());
        mod->SetUniqueName(name.c_str());
    }

    JobCGraphVtx* v = 0;
    for (int ind = 0; (v=this->GetAttached(ind))!=0; ++ind) {
        JobCPath* p2 = dynamic_cast<JobCPath*>(v);
        p2->SetUniquePaths(path.c_str());
    }
}

---