#include <DaqSource.h>

Inheritance diagram for edm::DaqSource:

Public Member Functions
	DaqSource (const ParameterSet &pset, const InputSourceDescription &desc)

virtual	~DaqSource ()

Public Member Functions inherited from edm::InputSource
boost::shared_ptr < ActivityRegistry >	actReg () const
	Accessor for Activity Registry. More...

void	closeFile (boost::shared_ptr< FileBlock >, bool cleaningUpAfterException)
	close current file More...

void	doBeginJob ()
	Called by framework at beginning of job. More...

void	doBeginLumi (LuminosityBlockPrincipal &lbp)
	Called by framework at beginning of lumi block. More...

void	doBeginRun (RunPrincipal &rp)
	Called by framework at beginning of run. More...

void	doEndJob ()
	Called by framework at end of job. More...

void	doEndLumi (LuminosityBlockPrincipal &lbp, bool cleaningUpAfterException)
	Called by framework at end of lumi block. More...

void	doEndRun (RunPrincipal &rp, bool cleaningUpAfterException)
	Called by framework at end of run. More...

void	doPostForkReacquireResources (boost::shared_ptr< multicore::MessageReceiverForSource >)

void	doPreForkReleaseResources ()
	Called by the framework before forking the process. More...

ProcessingController::ForwardState	forwardState () const

bool	goToEvent (EventID const &eventID)

	InputSource (ParameterSet const &, InputSourceDescription const &)
	Constructor. More...

void	issueReports (EventID const &eventID)
	issue an event report More...

LuminosityBlockNumber_t	luminosityBlock () const
	Accessor for current luminosity block number. More...

boost::shared_ptr < LuminosityBlockAuxiliary >	luminosityBlockAuxiliary () const
	Called by the framework to merge or insert lumi in principal cache. More...

int	markLumi ()
	Mark lumi as read. More...

int	markRun ()
	Mark run as read. More...

int	maxEvents () const

int	maxLuminosityBlocks () const

ModuleDescription const &	moduleDescription () const
	Accessor for 'module' description. More...

ItemType	nextItemType ()

bool	primary () const
	Accessor for primary input source flag. More...

ProcessConfiguration const &	processConfiguration () const
	Accessor for Process Configuration. More...

std::string const &	processGUID () const
	Accessor for global process identifier. More...

ProcessingMode	processingMode () const
	RunsLumisAndEvents (default), RunsAndLumis, or Runs. More...

boost::shared_ptr < ProductRegistry const >	productRegistry () const
	Accessor for product registry. More...

bool	randomAccess () const

void	readAndCacheLumi (bool merge, HistoryAppender &historyAppender)
	Read next luminosity block. More...

void	readAndCacheRun (bool merge, HistoryAppender &historyAppender)
	Read next run. More...

EventPrincipal *	readEvent (boost::shared_ptr< LuminosityBlockPrincipal > lbCache)

EventPrincipal *	readEvent (EventID const &)
	Read a specific event. More...

boost::shared_ptr< FileBlock >	readFile ()
	Read next file. More...

boost::shared_ptr < LuminosityBlockAuxiliary >	readLuminosityBlockAuxiliary ()
	Read next luminosity block Auxilary. More...

boost::shared_ptr< RunAuxiliary >	readRunAuxiliary ()
	Read next run Auxiliary. More...

ProcessHistoryID const &	reducedProcessHistoryID () const

void	registerProducts ()
	Register any produced products. More...

int	remainingEvents () const

int	remainingLuminosityBlocks () const

void	repeat ()
	Reset the remaining number of events/lumis to the maximum number. More...

ProcessingController::ReverseState	reverseState () const

void	rewind ()
	Begin again at the first event. More...

RunNumber_t	run () const
	Accessor for current run number. More...

boost::shared_ptr< RunAuxiliary >	runAuxiliary () const
	Called by the framework to merge or insert run in principal cache. More...

void	setLuminosityBlockNumber_t (LuminosityBlockNumber_t lb)
	Set the luminosity block ID. More...

void	setRunNumber (RunNumber_t r)
	Set the run number. More...

void	skipEvents (int offset)

Timestamp const &	timestamp () const
	Accessor for the current time, as seen by the input source. More...

virtual	~InputSource ()
	Destructor. More...

Private Member Functions
virtual void	closeBackDoor ()

void	defaultWebPage (xgi::Input in, xgi::Output out)

int	doMyBeginRun ()

virtual ItemType	getNextItemType ()

virtual void	openBackDoor (unsigned int, bool *)

virtual void	publish (xdata::InfoSpace *)

virtual void	publishForkInfo (evf::moduleweb::ForkInfoObj *forkInfoObj)

virtual void	publishToXmas (xdata::InfoSpace *)

virtual EventPrincipal *	readEvent_ ()

virtual EventPrincipal *	readIt (EventID const &eventID)

virtual boost::shared_ptr < LuminosityBlockAuxiliary >	readLuminosityBlockAuxiliary_ ()

virtual boost::shared_ptr < RunAuxiliary >	readRunAuxiliary_ ()

virtual void	setLumi (LuminosityBlockNumber_t lb)

virtual void	setRun (RunNumber_t r)

virtual void	signalWaitingThreadAndBlock ()

virtual void	skip (int offset)

Private Member Functions inherited from evf::ModuleWeb
	ModuleWeb (const std::string &)

virtual	~ModuleWeb ()

Private Attributes
bool	alignLsToLast_

bool	beginRunTiming_

pthread_cond_t	cond_

int	count

DaqProvenanceHelper	daqProvenanceHelper_

bool	eventCached_

unsigned int	eventCounter_

bool	fakeLSid_

evf::moduleweb::ForkInfoObj *	forkInfo_

bool	goToStopping

bool	immediateStop

xdata::InfoSpace *	is_

bool	keepUsingPsidFromTrigger_

xdata::UnsignedInteger32 *	lastLumiPrescaleIndex_

xdata::UnsignedInteger32 *	lastLumiUsingEol_

xdata::Boolean *	lsTimedOut_

xdata::Boolean *	lsToBeRecovered_

LuminosityBlockNumber_t	luminosityBlockNumber_

xdata::UnsignedInteger32 *	lumiSectionIndex_

unsigned int	lumiSegmentSizeInEvents_

xdata::InfoSpace *	mis_

pthread_mutex_t	mutex_

bool	newLumi_

bool	newRun_

bool	noMoreEvents_

ProcessHistoryID	phid_

xdata::UnsignedInteger32 *	prescaleSetIndex_

DaqBaseReader *	reader_

bool	runFork_

RunNumber_t	runNumber_

pthread_mutex_t	signal_lock_

unsigned int	thisEventLSid

timeval	tvStat_

bool	useEventCounter_

Private Attributes inherited from evf::ModuleWeb
std::string	moduleName_

Additional Inherited Members
Public Types inherited from edm::InputSource
enum	ItemType { IsInvalid, IsStop, IsFile, IsRun, IsLumi, IsEvent, IsRepeat }

enum	ProcessingMode { Runs, RunsAndLumis, RunsLumisAndEvents }

typedef ProductRegistryHelper::TypeLabelList	TypeLabelList

Static Public Member Functions inherited from edm::InputSource
static const std::string &	baseType ()

static void	fillDescription (ParameterSetDescription &desc)

static void	fillDescriptions (ConfigurationDescriptions &descriptions)

static void	prevalidate (ConfigurationDescriptions &)

Protected Member Functions inherited from edm::InputSource
void	decreaseRemainingEventsBy (int iSkipped)

EventPrincipal *	eventPrincipalCache ()

boost::shared_ptr < LuminosityBlockPrincipal > const	luminosityBlockPrincipal () const

PrincipalCache const &	principalCache () const

PrincipalCache &	principalCache ()

ProductRegistry &	productRegistryUpdate () const

void	reset () const

void	resetLuminosityBlockAuxiliary () const

void	resetRunAuxiliary () const

boost::shared_ptr < RunPrincipal > const	runPrincipal () const

void	setLuminosityBlockAuxiliary (LuminosityBlockAuxiliary *lbp)

void	setRunAuxiliary (RunAuxiliary *rp)

void	setTimestamp (Timestamp const &theTime)
	To set the current time, as seen by the input source. More...

ItemType	state () const

Detailed Description

Definition at line 36 of file DaqSource.h.

Constructor & Destructor Documentation

DaqSource::DaqSource	(	const ParameterSet &	pset,
		const InputSourceDescription &	desc
	)

explicit

Definition at line 56 of file DaqSource.cc.

References edm::Timestamp::beginOfTime(), cms::Exception::category(), cond_, count, edm::DaqProvenanceHelper::daqInit(), daqProvenanceHelper_, alignCSCRings::e, reco::get(), edm::ParameterSet::getUntrackedParameter(), mutex_, phid_, edm::InputSource::productRegistryUpdate(), matplotRender::reader, reader_, runNumber_, DaqBaseReader::setRunNumber(), edm::InputSource::setTimestamp(), and signal_lock_.

     : InputSource(pset,desc)
     , evf::ModuleWeb("DaqSource")
     , reader_(0)
     , lumiSegmentSizeInEvents_(pset.getUntrackedParameter<unsigned int>("evtsPerLS",0))
     , useEventCounter_(pset.getUntrackedParameter<bool>("useEventCounter",false))
     , eventCounter_(0)
     , keepUsingPsidFromTrigger_(pset.getUntrackedParameter<bool>("keepUsingPsidFromTrigger",false))
     , fakeLSid_(lumiSegmentSizeInEvents_ != 0)
     , runNumber_(RunID::firstValidRun().run())
     , luminosityBlockNumber_(LuminosityBlockID::firstValidLuminosityBlock().luminosityBlock())
     , daqProvenanceHelper_(TypeID(typeid(FEDRawDataCollection)))
     , noMoreEvents_(false)
     , newRun_(true)
     , newLumi_(true)
     , eventCached_(false)
     , alignLsToLast_(false)
     , is_(0)
     , mis_(0)
     , thisEventLSid(0)
     , goToStopping(false)
     , immediateStop(false)
     , forkInfo_(nullptr)
     , runFork_(false)
     , beginRunTiming_(false)
   {
     count = 0;
     pthread_mutex_init(&mutex_,0);
     pthread_mutex_init(&signal_lock_,0);
     pthread_cond_init(&cond_,0);
 
 
     setTimestamp(Timestamp::beginOfTime());
     
     // Instantiate the requested data source
     std::string reader = pset.getUntrackedParameter<std::string>("readerPluginName");
     
     try{
       reader_=
         DaqReaderPluginFactory::get()->create(reader,
                         pset.getUntrackedParameter<ParameterSet>("readerPset"));
       reader_->setRunNumber(runNumber_);
     }
     catch(edm::Exception &e) {
       if(e.category() == "Configuration" && reader_ == 0) {
     reader_ = DaqReaderPluginFactoryU::get()->create(reader);
     if(reader_ == 0) throw;
     else reader_->setRunNumber(runNumber_);
       }
       else {
         throw;
       }
     }
 
    // Initialize metadata, and save the process history ID for use every event.
    phid_ = daqProvenanceHelper_.daqInit(productRegistryUpdate());
 
   }

DaqSource::~DaqSource ( )

virtual

Definition at line 117 of file DaqSource.cc.

References reader_.

                         {
     delete reader_;
   }

Member Function Documentation

void DaqSource::closeBackDoor ( )

privatevirtual

Reimplemented from evf::ModuleWeb.

Definition at line 634 of file DaqSource.cc.

References cond_, count, lsTimedOut_, mutex_, and signal_lock_.

   {
     count--;
     pthread_cond_signal(&cond_);
     pthread_mutex_unlock(&mutex_);
     pthread_mutex_lock(&signal_lock_);
     lsTimedOut_->value_ = false; 
   }

void DaqSource::defaultWebPage	(	xgi::Input *	in,
		xgi::Output *	out
	)

privatevirtual

Reimplemented from evf::ModuleWeb.

Definition at line 656 of file DaqSource.cc.

References alignCSCRings::e, cppFunctionSkipper::exception, edm::first(), goToStopping, LaserDQM_cfg::input, evf::ModuleWeb::moduleName_, scaleCards::path, pos, o2o::query, and edm::second().

   {
       std::string path;
       std::string urn;
       std::string mname;
       std::string query;
       std::string original_referrer_;
       try 
     {
       cgicc::Cgicc cgi(in);
       if ( xgi::Utils::hasFormElement(cgi,"gotostopping") )
         {
           goToStopping=true;
         }
       if ( xgi::Utils::hasFormElement(cgi,"module") )
         mname = xgi::Utils::getFormElement(cgi, "module")->getValue();
       cgicc::CgiEnvironment cgie(in);
       if(original_referrer_ == "")
         original_referrer_ = cgie.getReferrer();
       path = cgie.getPathInfo();
       query = cgie.getQueryString();
     }
       catch (const std::exception & e) 
     {
       // don't care if it did not work
     }
 
       using std::endl;
       *out << "<html>"                                                   << endl;
       *out << "<head>"                                                   << endl;
 
 
       *out << "<STYLE type=\"text/css\"> #T1 {border-width: 2px; border: solid blue; text-align: center} </STYLE> "                                      << endl; 
       *out << "<link type=\"text/css\" rel=\"stylesheet\"";
       *out << " href=\"/" <<  urn
        << "/styles.css\"/>"                   << endl;
 
       *out << "<title>" << moduleName_
        << " MAIN</title>"                                            << endl;
 
       *out << "</head>"                                                  << endl;
       *out << "<body onload=\"loadXMLDoc()\">"                           << endl;
       *out << "<table border=\"0\" width=\"100%\">"                      << endl;
       *out << "<tr>"                                                     << endl;
       *out << "  <td align=\"left\">"                                    << endl;
       *out << "    <img"                                                 << endl;
       *out << "     align=\"middle\""                                    << endl;
       *out << "     src=\"/evf/images/bugicon.jpg\""                     << endl;
       *out << "     alt=\"main\""                                        << endl;
       *out << "     width=\"90\""                                        << endl;
       *out << "     height=\"64\""                                       << endl;
       *out << "     border=\"\"/>"                                       << endl;
       *out << "    <b>"                                                  << endl;
       *out <<             moduleName_                                    << endl;
       *out << "    </b>"                                                 << endl;
       *out << "  </td>"                                                  << endl;
       *out << "  <td width=\"32\">"                                      << endl;
       *out << "    <a href=\"/urn:xdaq-application:lid=3\">"             << endl;
       *out << "      <img"                                               << endl;
       *out << "       align=\"middle\""                                  << endl;
       *out << "       src=\"/hyperdaq/images/HyperDAQ.jpg\""             << endl;
       *out << "       alt=\"HyperDAQ\""                                  << endl;
       *out << "       width=\"32\""                                      << endl;
       *out << "       height=\"32\""                                     << endl;
       *out << "       border=\"\"/>"                                     << endl;
       *out << "    </a>"                                                 << endl;
       *out << "  </td>"                                                  << endl;
       *out << "  <td width=\"32\">"                                      << endl;
       *out << "  </td>"                                                  << endl;
       *out << "  <td width=\"32\">"                                      << endl;
       *out << "    <a href=\"" << original_referrer_  << "\">"           << endl;
       *out << "      <img"                                               << endl;
       *out << "       align=\"middle\""                                  << endl;
       *out << "       src=\"/evf/images/spoticon.jpg\""          << endl;
       *out << "       alt=\"main\""                                      << endl;
       *out << "       width=\"32\""                                      << endl;
       *out << "       height=\"32\""                                     << endl;
       *out << "       border=\"\"/>"                                     << endl;
       *out << "    </a>"                                                 << endl;
       *out << "  </td>"                                                  << endl;
       *out << "</tr>"                                                    << endl;
       *out << "</table>"                                                 << endl;
 
       *out << "<hr/>"                                                    << endl;
   
       *out << cgicc::form().set("method","GET").set("action", path ) 
        << std::endl;
       boost::char_separator<char> sep("&");
       boost::tokenizer<boost::char_separator<char> > tokens(query, sep);
       for (boost::tokenizer<boost::char_separator<char> >::iterator tok_iter = tokens.begin();
        tok_iter != tokens.end(); ++tok_iter){
     size_t pos = (*tok_iter).find_first_of("=");
     if(pos != std::string::npos){
       std::string first  = (*tok_iter).substr(0    ,                        pos);
       std::string second = (*tok_iter).substr(pos+1, (*tok_iter).length()-pos-1);
       *out << cgicc::input().set("type","hidden").set("name",first).set("value", second) 
            << std::endl;
     }
       }
 
       *out << cgicc::input().set("type","hidden").set("name","gotostopping").set("value","true")
        << std::endl;
       *out << cgicc::input().set("type","submit").set("value","Go To Stopping")          << std::endl;
       *out << cgicc::form()                        << std::endl;  
 
       *out << "</body>"                                                  << endl;
       *out << "</html>"                                                  << endl;
   }

int DaqSource::doMyBeginRun ( )

private

Definition at line 134 of file DaqSource.cc.

Referenced by getNextItemType().

                               {
 
     if (forkInfo_) {
       while (!immediateStop) {
     //queue new run to Framework (causes EP beginRun to be executed)
     if (newRun_) {
       beginRunTiming_=true;
       gettimeofday(&tvStat_, NULL);
       return 2;
     }
     //measure time in fwk beginRun
     if (beginRunTiming_) {
       timeval tsTmp;
       gettimeofday(&tsTmp,NULL);
       long tusecs = (tsTmp.tv_sec-tvStat_.tv_sec)*1000000 + tsTmp.tv_usec - tvStat_.tv_usec;
       double tsecs = ((double)(tusecs/10000))/100.;
       std::cout << "DaqSource: FWK beginRun elapsed time: " << tsecs << " seconds in master EP"<< std::endl;
       edm::LogInfo("DaqSource") << "FWK beginRun elapsed time: " << tsecs << " seconds in master EP";
       beginRunTiming_=false;
       usleep(10000);//short sleep before fork
     }
     //first or new run init
     if (forkInfo_->forkParams.isMaster==-1) {
       forkInfo_->lock();//keeping it locked during init!
       forkInfo_->forkHandler(forkInfo_->fuAddr); //fork all slaves
         }
     if (forkInfo_->forkParams.isMaster==-1) {
       forkInfo_->unlock();
       std::cout << "ERROR (DaqSource): not notified to be either in  master or slave process after fork" << std::endl;
       return -2;
     }
 
     //slave process after fork: exit all this
     if (forkInfo_->forkParams.isMaster==0) {
       forkInfo_->unlock();
       return 1;
     }
 
     //master process after fork:
     if (forkInfo_->forkParams.isMaster==1) {
         forkInfo_->unlock();
         int slotToRestart=-1;
         sem_wait(forkInfo_->control_sem_);
         forkInfo_->lock();
 
         //got unblocked due to next run
         if (forkInfo_->forkParams.isMaster==-1) {
                 forkInfo_->unlock();
             continue; // check here for newRun_?
             }
         //check if asked to stop
         immediateStop=forkInfo_->stopCondition;
         if (immediateStop) {
           forkInfo_->receivedStop_=true;
           break;
         }
         
         //check if asked to restart
         slotToRestart = forkInfo_->forkParams.slotId;
 
       if (slotToRestart==-1 && forkInfo_->forkParams.restart==0) {
         //this will deal with spurious semaphore signals when slave is killed
         forkInfo_->unlock();
         continue;
       }
       //restart single slave
       forkInfo_->forkHandler(forkInfo_->fuAddr);
     }
       }
       //loop exit
       forkInfo_->unlock();
       return 0;
     }
     return -1; //no forkInfo_
   }

InputSource::ItemType DaqSource::getNextItemType ( )

privatevirtual

Implements edm::InputSource.

Definition at line 213 of file DaqSource.cc.

                              {
     //    std::cout << getpid() << " enter getNextItemType " << std::endl;
     if (runFork_) {
       runFork_=false;
       int queueNext = doMyBeginRun();
       //check if new run (requires returning IsRun once)
       if (queueNext == 2) runFork_=true;
     }
 
     //get initial time before beginRun (used with old forking)
     if (!forkInfo_ && newRun_) {
       beginRunTiming_=true;
       gettimeofday(&tvStat_, NULL);
     }
 
     if (immediateStop) return IsStop;
 
     // --------------
     if(goToStopping){noMoreEvents_ = true; goToStopping=false;}
     if (noMoreEvents_) {
       pthread_mutex_lock(&mutex_);
       pthread_cond_signal(&cond_);
       pthread_mutex_unlock(&mutex_);
       return IsStop;
     }
     if (newRun_) {
       return IsRun;
     }
 
     //calculate and print the beginRun the timing
     if (beginRunTiming_) {
       timeval tsTmp;
       gettimeofday(&tsTmp,NULL);
       long tusecs = (tsTmp.tv_sec-tvStat_.tv_sec)*1000000 + tsTmp.tv_usec - tvStat_.tv_usec;
       double tsecs = ((double)(tusecs/10000))/100.;
       std::cout << "DaqSource (slave pid "<< getpid() << " ): FWK beginRun elapsed time: " 
         << tsecs << " seconds "<< std::endl;
       edm::LogInfo("DaqSource") << "DaqSource (slave pid "<< getpid() << " ): FWK beginRun elapsed time: " 
         << tsecs << " seconds ";
       beginRunTiming_=false;
     }
 
     if (newLumi_ && luminosityBlockAuxiliary()) {
       //      std::cout << "newLumi & lumiblock valid " << std::endl;
       return IsLumi;
     }
     if (alignLsToLast_) { //here we are recovering from a gap in Ls number so an event may already be cached but 
       // we hold onto it until we have issued all the necessary endLumi/beginLumi
       //       std::cout << getpid() << "alignLsToLast was set and ls number is " 
       //        << luminosityBlockNumber_ << " before signaling" << std::endl;
       signalWaitingThreadAndBlock();
       luminosityBlockNumber_++;
       //       std::cout << getpid() << "alignLsToLast signaled and incremented " 
       //        << luminosityBlockNumber_ << " eventcached " 
       //        << eventCached_ << std::endl;
       newLumi_ = true;
       lumiSectionIndex_->value_ = luminosityBlockNumber_;
       resetLuminosityBlockAuxiliary();
       if(luminosityBlockNumber_ == thisEventLSid+1) 
       {
         alignLsToLast_ = false;
       }
       if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != luminosityBlockNumber_) {
     setLuminosityBlockAuxiliary(new LuminosityBlockAuxiliary(
           runNumber_, luminosityBlockNumber_, timestamp(), Timestamp::invalidTimestamp()));
     luminosityBlockAuxiliary()->setProcessHistoryID(phid_);
 
     //      std::cout << "nextItemType: dealt with new lumi block principal, retval is " << retval << std::endl;
       }
       return IsLumi;
     }
     if (eventCached_) {
       //      std::cout << "read event already cached " << std::endl;
       return IsEvent;
     }
     if(reader_ == 0) {
       throw edm::Exception(errors::LogicError)
       << "DaqSource is used without a reader. Check your configuration !";
     }
     EventID eventId;
     TimeValue_t time = 0LL;
     timeval stv;
     gettimeofday(&stv,0);
     time = stv.tv_sec;
     time = (time << 32) + stv.tv_usec;
     Timestamp tstamp(time);
 
     int bunchCrossing = EventAuxiliary::invalidBunchXing;
     int orbitNumber   = EventAuxiliary::invalidBunchXing;
     
     // pass a 0 pointer to fillRawData()!
     FEDRawDataCollection* fedCollection(0);
 
     edm::EventAuxiliary::ExperimentType evttype = EventAuxiliary::Undefined;
   
     // let reader_ fill the fedCollection 
     int retval = reader_->fillRawData(eventId, tstamp, fedCollection);
     if(retval==0) {
       // fillRawData() failed, clean up the fedCollection in case it was allocated!
       if (0 != fedCollection) delete fedCollection;
       noMoreEvents_ = true;
       pthread_mutex_lock(&mutex_);
       pthread_cond_signal(&cond_);
       pthread_mutex_unlock(&mutex_);
       return IsStop;
     }
     else if(retval<0)
       {
  
     unsigned int nextLsFromSignal = (-1)*retval+1;
 //  std::cout << getpid() << "::got end-of-lumi for " << (-1)*retval
 //        << " was " << luminosityBlockNumber_ << std::endl;
     if(luminosityBlockNumber_ == (nextLsFromSignal-1) )
       {
         lastLumiUsingEol_->value_ = nextLsFromSignal;
         if(lsToBeRecovered_->value_){
 //        std::cout << getpid() << "eol::recover ls::for " << (-1)*retval << std::endl;
           signalWaitingThreadAndBlock();
           luminosityBlockNumber_++;
           newLumi_ = true;
           lumiSectionIndex_->value_ = luminosityBlockNumber_;
           resetLuminosityBlockAuxiliary();
           thisEventLSid = nextLsFromSignal - 1;
           if(luminosityBlockNumber_ != thisEventLSid+1) 
         alignLsToLast_ = true;
           //          std::cout << getpid() << "eol::::alignLsToLast_ " << alignLsToLast_ << std::endl;
         }
         else{
           //          std::cout << getpid() << "eol::realign ls::for " << (-1)*retval << std::endl;
           luminosityBlockNumber_ = nextLsFromSignal;
           newLumi_ = true;
           lumiSectionIndex_->value_ = luminosityBlockNumber_;
           resetLuminosityBlockAuxiliary();
         }
       }
     else {
       if(nextLsFromSignal >(luminosityBlockNumber_+100) ) {
         edm::LogError("DaqSource") << "Got EOL event with value " << retval 
                        << " nextLS would be " << nextLsFromSignal 
                        << " while we expected " << luminosityBlockNumber_+1 << " - disregarding... ";
       }
       if (nextLsFromSignal > luminosityBlockNumber_+2) //recover on delta > 2
       {
           lastLumiUsingEol_->value_ = nextLsFromSignal;
               thisEventLSid=nextLsFromSignal-1;//set new LS
           signalWaitingThreadAndBlock();
           luminosityBlockNumber_++;
           newLumi_ = true;
           lumiSectionIndex_->value_ = luminosityBlockNumber_;
           alignLsToLast_ = true;
 
           //set new lumi block
           resetLuminosityBlockAuxiliary();
           setLuminosityBlockAuxiliary(new LuminosityBlockAuxiliary(
             runNumber_, luminosityBlockNumber_, timestamp(), Timestamp::invalidTimestamp()));
           luminosityBlockAuxiliary()->setProcessHistoryID(phid_);
       }
 
     }
     //  else
     //    std::cout << getpid() << "::skipping end-of-lumi for " << (-1)*retval << std::endl;
       }
     else
       {
     if (eventId.event() == 0) {
       throw edm::Exception(errors::LogicError)
         << "The reader used with DaqSource has returned an invalid (zero) event number!\n"
         << "Event numbers must begin at 1, not 0.";
     }
     EventSourceSentry(*this);
     setTimestamp(tstamp);
     
     unsigned char *gtpFedAddr = fedCollection->FEDData(daqsource::gtpEvmId_).size()!=0 ? fedCollection->FEDData(daqsource::gtpEvmId_).data() : 0;
     uint32_t gtpsize = 0;
     if(gtpFedAddr !=0) gtpsize = fedCollection->FEDData(daqsource::gtpEvmId_).size();
     unsigned char *gtpeFedAddr = fedCollection->FEDData(daqsource::gtpeId_).size()!=0 ? fedCollection->FEDData(daqsource::gtpeId_).data() : 0; 
 
     unsigned int nextFakeLs = 0;
     eventCounter_++;
     if (fakeLSid_)
         evttype =  edm::EventAuxiliary::PhysicsTrigger; 
     if(fakeLSid_ && luminosityBlockNumber_ != 
        (nextFakeLs = useEventCounter_ ? ((eventCounter_-1)/lumiSegmentSizeInEvents_ + 1) :
         ((eventId.event() - 1)/lumiSegmentSizeInEvents_ + 1))) {
       lastLumiPrescaleIndex_->value_ = prescaleSetIndex_->value_;
       prescaleSetIndex_->value_ = 0; // since we do not know better but we want to be able to run
      
       if(luminosityBlockNumber_ == nextFakeLs-1)
         signalWaitingThreadAndBlock();
       luminosityBlockNumber_ = nextFakeLs;
       thisEventLSid = nextFakeLs-1;
       newLumi_ = true;
       lumiSectionIndex_->value_ = luminosityBlockNumber_;
       resetLuminosityBlockAuxiliary();
       if(keepUsingPsidFromTrigger_ && 
          gtpFedAddr!=0 && evf::evtn::evm_board_sense(gtpFedAddr,gtpsize)){
         prescaleSetIndex_->value_  = (evf::evtn::getfdlpsc(gtpFedAddr) & 0xffff);
       }   
     }
     else if(!fakeLSid_){ 
 
       if(gtpFedAddr!=0 && evf::evtn::evm_board_sense(gtpFedAddr,gtpsize)){
         lastLumiPrescaleIndex_->value_ = prescaleSetIndex_->value_;
         thisEventLSid = evf::evtn::getlbn(gtpFedAddr);
         prescaleSetIndex_->value_  = (evf::evtn::getfdlpsc(gtpFedAddr) & 0xffff);
         evttype =  edm::EventAuxiliary::ExperimentType(evf::evtn::getevtyp(gtpFedAddr));
         if(luminosityBlockNumber_ > (thisEventLSid + 1))
         {
           //late event,throw fwk exception
           std::ostringstream excptmsg;
           excptmsg << "DaqSource::event with late LS (" << thisEventLSid + 1 << ")received.";
               throw edm::Exception(errors::LogicError,excptmsg.str());
         }
         if(luminosityBlockNumber_ != (thisEventLSid + 1)){
           // we got here in a running process and some Ls might have been skipped so set the flag, 
           // increase by one, check and if appropriate set the flag then continue
           if(lsToBeRecovered_->value_){
         //      std::cout << getpid() << "eve::recover ls::for " << thisEventLSid << std::endl;
         signalWaitingThreadAndBlock();
         luminosityBlockNumber_++;
         newLumi_ = true;
         lumiSectionIndex_->value_ = luminosityBlockNumber_;
         resetLuminosityBlockAuxiliary();
         if(luminosityBlockNumber_ != thisEventLSid+1) alignLsToLast_ = true;
         //      std::cout << getpid() << "eve::::alignLsToLast_ " << alignLsToLast_ << std::endl;
           }
           else{ // we got here because the process was restarted. just realign the ls id and proceed with this event
         //      std::cout << getpid() << "eve::realign ls::for " << thisEventLSid << std::endl;
         luminosityBlockNumber_ = thisEventLSid + 1;
         newLumi_ = true;
         lumiSectionIndex_->value_ = luminosityBlockNumber_;
         resetLuminosityBlockAuxiliary();
         lsToBeRecovered_->value_ = true;
           }
         }
       }
       else if(gtpeFedAddr!=0 && evf::evtn::gtpe_board_sense(gtpeFedAddr)){
         lastLumiPrescaleIndex_->value_ = prescaleSetIndex_->value_;
         thisEventLSid = evf::evtn::gtpe_getlbn(gtpeFedAddr);
         prescaleSetIndex_->value_ = 0; //waiting to get a PS index from gtpe
         evttype =  edm::EventAuxiliary::PhysicsTrigger; 
         if(luminosityBlockNumber_ != (thisEventLSid + 1)){
           if(luminosityBlockNumber_ == thisEventLSid)
         signalWaitingThreadAndBlock();
           luminosityBlockNumber_ = thisEventLSid + 1;
           newLumi_ = true;
           lumiSectionIndex_->value_ = luminosityBlockNumber_;
           resetLuminosityBlockAuxiliary();
         }
       }
     }
     if(gtpFedAddr!=0 && evf::evtn::evm_board_sense(gtpFedAddr,gtpsize)){
       bunchCrossing =  int(evf::evtn::getfdlbx(gtpFedAddr));
       orbitNumber =  int(evf::evtn::getorbit(gtpFedAddr));
       TimeValue_t time = evf::evtn::getgpshigh(gtpFedAddr);
       time = (time << 32) + evf::evtn::getgpslow(gtpFedAddr);
       Timestamp tstamp(time);
       setTimestamp(tstamp);      
     }
     else if(gtpeFedAddr!=0 && evf::evtn::gtpe_board_sense(gtpeFedAddr)){
       bunchCrossing =  int(evf::evtn::gtpe_getbx(gtpeFedAddr));
       orbitNumber =  int(evf::evtn::gtpe_getorbit(gtpeFedAddr));
     }
       }    
           
     //    std::cout << "lumiblockaux = " << luminosityBlockAuxiliary() << std::endl;
     // If there is no luminosity block principal, make one.
     if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != luminosityBlockNumber_) {
       newLumi_ = true;
       setLuminosityBlockAuxiliary(new LuminosityBlockAuxiliary(
     runNumber_, luminosityBlockNumber_, timestamp(), Timestamp::invalidTimestamp()));
       luminosityBlockAuxiliary()->setProcessHistoryID(phid_);
 
       //      std::cout << "nextItemType: dealt with new lumi block principal, retval is " << retval << std::endl;
     }
     //    std::cout << "here retval = " << retval << std::endl;
     if(retval<0){
       //      std::cout << getpid() << " returning from getnextitem because retval < 0 - IsLumi "
       //        << IsLumi << std::endl;
       if(newLumi_) return IsLumi; else return getNextItemType();
     }
 
     // make a brand new event principal
     eventId = EventID(runNumber_,thisEventLSid+1, eventId.event());
     EventAuxiliary eventAux(eventId, processGUID(),
                 timestamp(),
                 true,
                 evttype,
                 bunchCrossing,
                 EventAuxiliary::invalidStoreNumber,
                 orbitNumber);
     eventAux.setProcessHistoryID(phid_);
     eventPrincipalCache()->fillEventPrincipal(eventAux, boost::shared_ptr<LuminosityBlockPrincipal>());
     eventCached_ = true;
     
     // have fedCollection managed by a std::auto_ptr<>
     std::auto_ptr<FEDRawDataCollection> bare_product(fedCollection);
 
     WrapperOwningHolder edp(new Wrapper<FEDRawDataCollection>(bare_product), Wrapper<FEDRawDataCollection>::getInterface());
     eventPrincipalCache()->put(daqProvenanceHelper_.constBranchDescription_, edp, daqProvenanceHelper_.dummyProvenance_);
 
 /*
     Event e(*eventPrincipalCache(), md_);
     // put the fed collection into the transient event store
     e.put(bare_product);
     // The commit is needed to complete the "put" transaction.
     e.commit_();
 */
     if (newLumi_) {
       return IsLumi;
     }
     return IsEvent;
   }

void DaqSource::openBackDoor	(	unsigned int	timeout_sec,
		bool *	running
	)

privatevirtual

Reimplemented from evf::ModuleWeb.

Definition at line 612 of file DaqSource.cc.

References CLOCK_REALTIME, cond_, count, lsTimedOut_, mutex_, NULL, and signal_lock_.

   {
     count++;
     if(count==2) throw;
     pthread_mutex_lock(&mutex_);
     if (running) *running=true;
     pthread_mutex_unlock(&signal_lock_);
     timespec ts;
 #if _POSIX_TIMERS > 0
     clock_gettime(CLOCK_REALTIME, &ts);
 #else
     struct timeval tv; 
     gettimeofday(&tv, NULL);
     ts.tv_sec = tv.tv_sec + 0;
     ts.tv_nsec = 0;
 #endif
     ts.tv_sec += timeout_sec;
 
     int rc = pthread_cond_timedwait(&cond_, &mutex_, &ts);
     if(rc == ETIMEDOUT) lsTimedOut_->value_ = true; 
   }

void DaqSource::publish ( xdata::InfoSpace * is )

privatevirtual

Implements evf::ModuleWeb.

Definition at line 597 of file DaqSource.cc.

References is_, lastLumiPrescaleIndex_, lastLumiUsingEol_, lsTimedOut_, lsToBeRecovered_, lumiSectionIndex_, and prescaleSetIndex_.

   {
     is_ = is;
     lumiSectionIndex_      = (xdata::UnsignedInteger32*)is_->find("lumiSectionIndex");
     prescaleSetIndex_      = (xdata::UnsignedInteger32*)is_->find("prescaleSetIndex");
     lastLumiPrescaleIndex_ = (xdata::UnsignedInteger32*)is_->find("lastLumiPrescaleIndex");
     lastLumiUsingEol_ = (xdata::UnsignedInteger32*)is_->find("lastLumiUsingEol");
     lsTimedOut_            = (xdata::Boolean*)is_->find("lsTimedOut");
     lsToBeRecovered_       = (xdata::Boolean*)is_->find("lsToBeRecovered");
   }

void DaqSource::publishForkInfo ( evf::moduleweb::ForkInfoObj * forkInfoObj )

privatevirtual

Reimplemented from evf::ModuleWeb.

Definition at line 121 of file DaqSource.cc.

References forkInfo_, immediateStop, noMoreEvents_, and runFork_.

                                                                        {
     forkInfo_ = forkInfoObj;
     runFork_=true;
     immediateStop=false;
     noMoreEvents_=false;
   }

void DaqSource::publishToXmas ( xdata::InfoSpace * is )

privatevirtual

Reimplemented from evf::ModuleWeb.

Definition at line 607 of file DaqSource.cc.

References mis_.

   {
     mis_ = is;
   }

EventPrincipal * DaqSource::readEvent_ ( )

privatevirtual

Implements edm::InputSource.

Definition at line 561 of file DaqSource.cc.

References eventCached_, edm::InputSource::eventPrincipalCache(), edm::InputSource::luminosityBlockPrincipal(), newLumi_, newRun_, noMoreEvents_, and edm::EventPrincipal::setLuminosityBlockPrincipal().

                         {
     //    std::cout << "assert not newRun " << std::endl;
     assert(!newRun_);
     //    std::cout << "assert not newLumi " << std::endl;
     assert(!newLumi_);
     //    std::cout << "assert not noMoreEvents " << std::endl;
     assert(!noMoreEvents_);
     //    std::cout << "assert eventCached " << std::endl;
     assert(eventCached_);
     //    std::cout << "asserts done " << std::endl;
     eventCached_ = false;
     eventPrincipalCache()->setLuminosityBlockPrincipal(luminosityBlockPrincipal());
     return eventPrincipalCache();
   }

EventPrincipal * DaqSource::readIt ( EventID const & eventID )

privatevirtual

Reimplemented from edm::InputSource.

Definition at line 584 of file DaqSource.cc.

References edm::hlt::Exception, and edm::errors::LogicError.

                                   {
       throw edm::Exception(errors::LogicError,"DaqSource::readIt(EventID const& eventID)")
         << "Random access read cannot be used for DaqSource.\n"
         << "Contact a Framework developer.\n";
   }

boost::shared_ptr< LuminosityBlockAuxiliary > DaqSource::readLuminosityBlockAuxiliary_ ( )

privatevirtual

Implements edm::InputSource.

Definition at line 549 of file DaqSource.cc.

References edm::InputSource::luminosityBlockAuxiliary(), newLumi_, newRun_, and noMoreEvents_.

                                            {
     assert(!newRun_);
     assert(newLumi_);
     assert(!noMoreEvents_);
     assert(luminosityBlockAuxiliary());
     //assert(eventCached_); //the event may or may not be cached - rely on 
     // the call to getNextItemType to detect that.
     newLumi_ = false;
     return luminosityBlockAuxiliary();
   }

boost::shared_ptr< RunAuxiliary > DaqSource::readRunAuxiliary_ ( )

privatevirtual

Implements edm::InputSource.

Definition at line 539 of file DaqSource.cc.

References edm::Timestamp::invalidTimestamp(), newRun_, noMoreEvents_, phid_, runNumber_, and edm::InputSource::timestamp().

                                {
     assert(newRun_);
     assert(!noMoreEvents_);
     newRun_ = false;
     boost::shared_ptr<RunAuxiliary> ra(new RunAuxiliary(runNumber_, timestamp(), Timestamp::invalidTimestamp()));
     ra->setProcessHistoryID(phid_);
     return ra;
   }

void DaqSource::setLumi ( LuminosityBlockNumber_t lb )

privatevirtual

Reimplemented from edm::InputSource.

Definition at line 577 of file DaqSource.cc.

References edm::hlt::Exception, and edm::errors::LogicError.

                                             {
       throw edm::Exception(errors::LogicError,"DaqSource::setLumi(LuminosityBlockNumber_t lumiNumber)")
         << "The luminosity block number cannot be set externally for DaqSource.\n"
         << "Contact a Framework developer.\n";
   }

void DaqSource::setRun ( RunNumber_t r )

privatevirtual

Reimplemented from edm::InputSource.

Definition at line 528 of file DaqSource.cc.

References eventCached_, newLumi_, newRun_, noMoreEvents_, alignCSCRings::r, reader_, edm::InputSource::reset(), edm::InputSource::resetLuminosityBlockAuxiliary(), runNumber_, and DaqBaseReader::setRunNumber().

                                  {
     assert(!eventCached_);
     reset();
     newRun_ = newLumi_ = true;
     runNumber_ = r;
     if (reader_) reader_->setRunNumber(runNumber_);
     noMoreEvents_ = false;
     resetLuminosityBlockAuxiliary();
   }

void DaqSource::signalWaitingThreadAndBlock ( )

privatevirtual

Definition at line 643 of file DaqSource.cc.

References cond_, mutex_, and signal_lock_.

Referenced by getNextItemType().

   {
     pthread_mutex_lock(&signal_lock_);
     pthread_mutex_lock(&mutex_);
     pthread_mutex_unlock(&signal_lock_);
     //    std::cout << getpid() << " DS::signal from evloop " << std::endl;
     pthread_cond_signal(&cond_);
     //    std::cout << getpid() << " DS::go to wait for scalers wl " << std::endl;
     pthread_cond_wait(&cond_, &mutex_);
     pthread_mutex_unlock(&mutex_);
     ::usleep(1000);//allow other thread to lock
   }  

void DaqSource::skip ( int offset )

privatevirtual

Reimplemented from edm::InputSource.

Definition at line 591 of file DaqSource.cc.

References edm::hlt::Exception, and edm::errors::LogicError.

                      {
       throw edm::Exception(errors::LogicError,"DaqSource::skip(int offset)")
         << "Random access skip cannot be used for DaqSource\n"
         << "Contact a Framework developer.\n";
   }