JobCGraphVtx.cxx

Go to the documentation of this file.

// $Id: JobCGraphVtx.cxx,v 1.6 2008/02/05 19:02:39 scavan Exp $
// 
// Implementation of a forward feeding directed graph
//
// messier@huhepl.harvard.edu
#include "JobCGraphVtx.h"
#include "MessageService/MsgService.h"
CVSID("$Id: JobCGraphVtx.cxx,v 1.6 2008/02/05 19:02:39 scavan Exp $");

ClassImp(JobCGraphVtx)

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

JobCGraphVtx::JobCGraphVtx() : 
  fNupStream(0), fNupStreamPassed(0), fNupStreamFailed(0) 
{ }

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

void JobCGraphVtx::Attach(JobCGraphVtx* JobCGraphVtx) 
{
  fDownStream.push_back(JobCGraphVtx);
  JobCGraphVtx->AddUpStream();
}

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

void JobCGraphVtx::Detach(JobCGraphVtx* /* JobCGraphVtx */) 
{
  // not implemented
  MSG("JobC",Msg::kWarning) << "Detach not implemented.\n";
}

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

void JobCGraphVtx::EvaluateGraph(JobCRecord* rec) 
{
  this->InitGraphEvaluate();
  this->EvaluateNode(rec);
}

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

void JobCGraphVtx::EvaluateNode(JobCRecord* rec) 
{
  // If we have streams feeding into this JobCGraphVtx check their status
  if (fNupStream) {
    // If all upstream nodes have failed just propagate that
    // information down the line
    if (fNupStreamFailed == fNupStream) {
      this->SendNotification(false);
      return;
    }
    // Check if we're still waiting for upstream nodes to finish
    if ((fNupStreamFailed+fNupStreamPassed)<fNupStream) {
      return;
    }
  }
  
  // Evaluate this node
  bool passed = this->EvalInput(rec);
  if (fDownStream.size()>0) {
    this->SendNotification(passed);
    vector<JobCGraphVtx*>::iterator itr(fDownStream.begin());
    vector<JobCGraphVtx*>::iterator itrEnd(fDownStream.end());
    for (;itr!=itrEnd;++itr) (*itr)->EvaluateNode(rec);
  }
}

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

void JobCGraphVtx::AddUpStream() { ++fNupStream; }

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

void JobCGraphVtx::ReceiveNotification(bool passfail)
{
  if (passfail) ++fNupStreamPassed;
  else          ++fNupStreamFailed;
}

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

void JobCGraphVtx::SendNotification(bool passfail) 
{
  vector<JobCGraphVtx*>::iterator itr(fDownStream.begin());
  vector<JobCGraphVtx*>::iterator itrEnd(fDownStream.end());
  for (;itr!=itrEnd;++itr) (*itr)->ReceiveNotification(passfail);
}

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

void JobCGraphVtx::InitGraphEvaluate() 
{
  fNupStreamPassed = 0;
  fNupStreamFailed = 0;
  this->DerivedInitGraphEvaluate();
  vector<JobCGraphVtx*>::iterator itr(fDownStream.begin());
  vector<JobCGraphVtx*>::iterator itrEnd(fDownStream.end());
  for (;itr!=itrEnd;++itr) (*itr)->InitGraphEvaluate();
}

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

JobCGraphVtx* JobCGraphVtx::GetAttached(unsigned int i) const
{ 
  //always true: assert(i>=0);
  if (i<fDownStream.size()) return fDownStream[i];
  return 0;
}

---