IoInputModule Class Reference

#include <IoInputModule.h>

Inheritance diagram for IoInputModule:

List of all members.

Public Member Functions
	IoInputModule ()
	~IoInputModule ()
void	BeginJob ()
void	EndJob ()
const Registry &	DefaultConfig () const
void	Config (const Registry &r)
JobCResult	Get ()
JobCResult	Next (int n=1)
JobCResult	Prev (int n=1)
JobCResult	GoTo (int run, int snarl, int searchDir=0)
JobCResult	GoTo (const VldContext &vld)
void	List (const char *streamlist="*") const
void	AddFile (const char *filename, const char *streamlist="*", int at=-1)
void	RemoveFile (const char *filename, const char *streamlist="*")
JobCResult	NextFile (int n=1, const char *streamlist="*")
JobCResult	PrevFile (int n=1, const char *streamlist="*")
JobCResult	GoToFile (int i, const char *streamlist="*")
JobCResult	GoToFile (const char *filename, const char *streamlist="*")
void	DefineStream (const char *stream, const char *tree)
void	Select (const char *stream, const char *select, bool isRequired=false)
void	SetSequenceMode (const char *stream, Per::ESequenceMode sequenceMode=Per::kKey)
void	SetPerOwnedDisabled (const char *stream, bool perowneddisabled=true)
void	SetTestMode (const char *stream, bool testMode)
void	SetWindow (const char *stream, double lower, double upper)
void	SetMaxFileRepeat (const char *stream, int numRepeat)
void	SetMeanMom (const char *stream, double mean)
void	SetPushRandom (const char *stream, bool setRandom)
void	SetRandomSeed (int rSeed)
Int_t	GetCurrentRun () const
Int_t	GetLastRun () const
const char *	GetCurrentFile (const char *streamname="*") const
const char *	GetLastFile (const char *streamname="*") const
Int_t	GetCurrentSnarl () const
Private Member Functions
void	LoadFilesFromCommandLine ()
int	ReadHeader ()
void	UpdateDDSConfig ()
void	UpdateFormatConfig ()
void	UpdateFileList ()
void	UpdateStreamConfig ()
int	OpenStreamItr ()
void	CloseStreamItr ()
JobCResult	Get (MomNavigator *)
Private Attributes
std::map< std::string, std::string >	fStreamSelectionMap
std::map< std::string, bool >	fStreamRequiredMap
std::map< std::string, std::string >	fStreamDefMap
std::map< std::string, Per::ESequenceMode >	fStreamSeqModeMap
std::map< std::string, bool >	fStreamPerOwnedDisabledMap
std::map< std::string, bool >	fStreamTestModeMap
std::map< std::string, std::pair< double, double > >	fStreamWindowMap
std::list< IoFileListItem >	fFileList
IoDataStreamItr *	fDataStreamItr
std::string	fFormat
std::string	fStreamList
std::map< std::string, int >	fStreamMaxRepeatMap
std::map< std::string, double >	fStreamMeanMap
std::map< std::string, bool >	fStreamPushRandomMap
int	fRandomSeed
std::string	fServer
int	fPort
int	fTimeOut
int	fDataSource
int	fKeepUpMode
int	fMaxSyncDelay
bool	fOffLine
int	fMaxRetry
int	fRetryDelay
DDS::EClientType	fClientType
string	fClientName
JobCResult	fStatus
int	fLastRun
int	fLastSnarl
int	fCurrentRun
int	fCurrentSnarl
std::map< std::string, std::string >	fCurrentFileMap
std::map< std::string, std::string >	fLastFileMap
std::string	fLastBeginFile
std::string	fLastEndFile
bool	fLoadedCommandLineFiles
TStopwatch	fStopwatch

---

Constructor & Destructor Documentation

IoInputModule::IoInputModule (   )

Definition at line 66 of file IoInputModule.cxx. References fStopwatch. : fDataStreamItr(0), fFormat(""), fStreamList(""), fServer(""), fPort(0), fTimeOut(0), fDataSource(0), fKeepUpMode(0), fMaxSyncDelay(0), fOffLine(false), fMaxRetry(0), fRetryDelay(1), fClientType(DDS::kUnknownClientType), fClientName(""), fStatus(JobCResult::kAOK), fLastRun(-1), fLastSnarl(-1), fCurrentRun(-1), fCurrentSnarl(-1), fLoadedCommandLineFiles(false) #ifdef SITE_HAS_SAM ,fsamProject(0) #endif { fStopwatch.Reset(); fStopwatch.Stop(); }

IoInputModule::~IoInputModule (   )

Definition at line 94 of file IoInputModule.cxx. References fDataStreamItr. { if ( fDataStreamItr ) { delete fDataStreamItr; fDataStreamItr = 0; } }

---

Member Function Documentation

void IoInputModule::AddFile (  const char *  filename, const char *  streamlist = "*", int  at = -1 )  [virtual]

Reimplemented from JobCInputModule. Definition at line 497 of file IoInputModule.cxx. References VldTimeStamp::AsString(), fDataStreamItr, fFileList, Registry::Get(), JobCModule::GetConfig(), IoFileListItem::GetFileStreamMap(), and MSG. Referenced by LoadFilesFromCommandLine(), IoInputModuleValidate::TestAdd(), and UpdateFileList(). { //====================================================================== // Add to the list of attached streams at the position "at". -1 = end // of list //====================================================================== // Find out by checking the format of the filepath whether this // is a SAM job. Format will be SAM:samdataset of SAM_FILE::file const char *s1 = "SAM"; if ( strstr(filepath,s1) != NULL ) { #ifdef SITE_HAS_SAM // SAM options const char* tmps; int tmpi; Registry& r = GetConfig(); if (r.Get("Station",tmps)) {fStation = tmps;} if (r.Get("SnapShotVers",tmpi)){fSnapShotVers = tmpi;} if (r.Get("WorkGroupName",tmps)) {fWorkGroupName = tmps;} if (r.Get("ApplicationName",tmps)) {fApplicationName = tmps;} if (r.Get("ApplicationVers",tmps)) {fApplicationVers = tmps;} if (r.Get("ProjectName",tmps)) {fProjectName = tmps;} if (r.Get("MaxNumberOfFiles",tmpi)) {fMaxNumberOfFiles = tmpi;} if (r.Get("StartNewProject",tmpi)) {fStartNewProject = tmpi;} // Decide whether we have a dataset or a single file // dataset will be SAM, file will be SAM_FILE // Get the samdataset or file from filepath std::string temp = filepath; size_t pos = temp.find(":")+1; std::string sam_access_type = temp.substr(0,pos-1); if (sam_access_type == "SAM") { std::string samdataset = temp.substr(pos,temp.length()-pos); std::string projectname; // If this is a new project then append time-stamp if (fStartNewProject == 1) { // Construct full project name including timestamp // Get timestamp as a string VldTimeStamp ts; std::string timestamp = ts.AsString("lc"); // Replace blank with "-" size_t pos = timestamp.find(" "); timestamp.replace(pos,1,"-"); // Replace : with - as DbServer does not like : in project names pos = timestamp.find(":"); while ( pos != string::npos ) { timestamp.replace(pos,1,"-"); pos = timestamp.find(":",pos+1); } // Append timestamp to fProjectName taken from registry fProjectName.append("-"); projectname = fProjectName+timestamp; } // Otherwise use project name that is supplied else if(fStartNewProject == 0) { projectname = fProjectName; } MSG("Io",Msg::kDebug) << "Sam Station " << fStation << " Snap Shot " << fSnapShotVers << " Work Group Name " << fWorkGroupName << " Application Name " << fApplicationName << " Application Version " << fApplicationVers << " Project Name " << projectname << endl; // Define snapshot version long snapshot; // Snapshot version = 0 means create New Snapshot if ( fSnapShotVers == 0 ) { snapshot = sam::SamProject::NewSnapshotVersion; } // Snapshot version < 0 means use last one created else if (fSnapShotVers < 0) { snapshot = sam::SamProject::LatestSnapshotVersion; } else if (fSnapShotVers > 0 ) { // Use specified SnapShot version snapshot = fSnapShotVers; } MSG("Io",Msg::kDebug) << "SnapShot Version " << snapshot << endl; if (fStartNewProject == 1) { // Create SAM project fsamProject = new sam::SamProject(projectname,fStation); // Start SAM project std::list<std::string> projectMasterArgList; try { MSG("Io",Msg::kInfo) << "Starting SAM Project " << projectname << " on station " << fStation << endl; fsamProject->startProject(fWorkGroupName,samdataset,snapshot, projectMasterArgList); } catch(const sam::SamProject::StartProjectRequestRejected& ex) { MSG("Io",Msg::kInfo) << "Rejected start SAM project request " << ex << endl; } } // Start SAM consumer to deliver files const int projectMasterTimeout(60); const std::string processDescription("Loon Analysis Process"); try{ sam::SamConsumer fsamConsumer(projectname,fStation,fWorkGroupName, fApplicationName,fApplicationVers, processDescription, fMaxNumberOfFiles, projectMasterTimeout); MSG("Io",Msg::kInfo) << "Started SAM Consumer" << endl; // Now get files. Format of returned files depends on whether // SAM cache is local disk, dcache disk or AFS file space std::map<std::string,std::string> filelist; map<std::string,std::string>::iterator fitr; int location; int length; int comp; std::string fileonly; std::string restOfPath; std::string afsroot("afsroot:"); try { while(true) { std::string filename = fsamConsumer.getFile().getFullFileName(); // The files need to be sorted as they come back in an undefined // order. Split them into a filename and the rest of the path. // Put them in a map and then iterate over key which is filename // - guarantees correct order. MSG("Io",Msg::kDebug) << "Filename " << filename << endl; location = filename.find_last_of("/"); length = filename.length(); fileonly = filename.substr(location+1,length-1); // Need to look for afsroot: at start of path. If it is there the // remove it and rest of path is AFS path. comp = filename.compare(0,8,afsroot); if (comp == 0 ) { restOfPath = filename.substr(8,location-8); } else { restOfPath = filename.substr(0,location); } restOfPath.append("/"); filelist.insert(make_pair(fileonly,restOfPath)); MSG("Io",Msg::kDebug) << "File Only " << fileonly << " Rest Of Path " << restOfPath << endl; // Release file fsamConsumer.releaseFile(); } } catch(const sam::SamConsumer::EndOfFileStreamReached& ex) { MSG("Io",Msg::kDebug) << "End of File Stream reached" << endl; } // Got all files. Now need to add them to file list. Iterate over map std::string sfile; const char *samfile = 0; for (fitr = filelist.begin(); fitr != filelist.end(); fitr++) { sfile = (fitr->second+fitr->first); samfile = sfile.data(); MSG("Io",Msg::kInfo) << "Adding File " << samfile << endl; // Add file to file list IoFileListItem iofile(samfile,at,streamlist); fFileList.push_back(iofile); } } catch(const sam::SamConsumer::InitializationError& ex) { MSG("Io",Msg::kInfo) << "Rejected start SAM Consumer request " << ex << endl; } if (fsamProject) { try { MSG("Io",Msg::kInfo) << "Requesting end of SAM project " << endl; fsamProject->endProject(); } catch(const sam::SamProject::EndProjectRequestRejected& ex) { MSG("Io",Msg::kInfo) << "SAM Project end request rejected "<< ex << endl; } catch(const sam::SamProject::EndProjectRequestFailed& ex) { MSG("Io",Msg::kInfo) << "SAM Project end request failed "<< ex << endl; } } } else if (sam_access_type == "SAM_FILE") { // User just wants a single file std::string samfile = temp.substr(pos,temp.length()-pos); // Use sam_locate to get pnfs path sam::LocationList samFiles; try { MSG("Io",Msg::kInfo) << "Locating file " << samfile << endl; samFiles = sam::locate(samfile); // Now need to translate pnfs path into dcache path // insert fnal.gov/usr/ after /pnfs/ samFiles[0].insert(6,"fnal.gov/usr/"); // There are two dcache ports so chose one at random TRandom3 rand(0); Double_t r = rand.Rndm(); if (r <= 0.5) { samFiles[0].insert(0,"dcap://fndca1.fnal.gov:24125"); } else if (r > 0.5) { samFiles[0].insert(0,"dcap://fndca1.fnal.gov:24136"); } // Finally append filename samFiles[0].append("/"); samFiles[0].append(samfile); MSG("Io",Msg::kInfo) << "Adding file " << samFiles[0].c_str() << " to input list" << endl; // Add file to file list IoFileListItem iofile(samFiles[0].c_str(),at,streamlist); fFileList.push_back(iofile); } catch(const sam::exception::DataFileNotFound& ex) { MSG("Io",Msg::kInfo) << ex << endl; } } #endif // End of ifdef SITE_HAS_SAM } else { // Add file to file list IoFileListItem iofile(filepath,at,streamlist); fFileList.push_back(iofile); if ( !fDataStreamItr ) return; // Add files to stream managed lists const IoFileListItem::FileStreamMap& filemap = iofile.GetFileStreamMap(); if ( at < 0 ) { IoFileListItem::FileStreamMapConstItr itr = filemap.begin(); for ( ; itr != filemap.end(); itr++ ) { std::string filename = itr -> first; std::string streamlist = itr -> second; fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str()); } } else { // Apply files in reverse to have first file in list inserted at pos At IoFileListItem::FileStreamMap::const_reverse_iterator itr=filemap.rbegin(); for ( ; itr != filemap.rend(); itr++ ) { std::string filename = itr -> first; std::string streamlist = itr -> second; fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str()); } } } }

void IoInputModule::BeginJob (   )  [virtual]

Implement for notification of begin of job Reimplemented from JobCModule. Definition at line 101 of file IoInputModule.cxx. References LoadFilesFromCommandLine(). { this->LoadFilesFromCommandLine(); // Delay opening files until first action (Next,Prev,etc.) call }

void IoInputModule::CloseStreamItr (   )  [private]

Definition at line 1516 of file IoInputModule.cxx. References fDataStreamItr, fStatus, IoDataStreamItr::GetFormat(), MSG, JobCResult::SetEndFile(), JobCResult::SetEndOfInputStream(), and JobCResult::SetEndRun(). Referenced by OpenStreamItr(), UpdateDDSConfig(), and UpdateFormatConfig(). { //====================================================================== // Close the currently openned stream //====================================================================== if (fDataStreamItr) { MSG("Io",Msg::kDebug) << "Close stream itr of format " << fDataStreamItr->GetFormat() << endl; delete fDataStreamItr; fDataStreamItr = 0; fStatus.SetEndRun(); fStatus.SetEndFile(); fStatus.SetEndOfInputStream(); } }

void IoInputModule::Config (  const Registry &  r  )  [virtual]

Return the actual configuration. If your module directly pulls its configuration from the fConfig Registry, you don't need to override this. Override if you have local config variables. Reimplemented from JobCModule. Definition at line 176 of file IoInputModule.cxx. References fClientName, fClientType, fDataSource, fFormat, fKeepUpMode, fMaxRetry, fMaxSyncDelay, fOffLine, fPort, fRetryDelay, fServer, fStreamList, fTimeOut, Registry::Get(), DDS::GetClientType(), DDS::GetDataSourceCode(), DDS::GetKeepUpCode(), MSG, UpdateDDSConfig(), UpdateFormatConfig(), and UpdateStreamConfig(). Referenced by IoInputModuleValidate::IoInputModuleValidate(). { //====================================================================== // Configure the module based on the contents of the registry r //====================================================================== const char* tmps; int tmpi; int tmpb; // bools MSG("Io",Msg::kDebug) << "Config IoInputModule with r=" << r << "\n"; // Input data stream configuration bool doFormatConfig = false; if (r.Get("Format", tmps)) { fFormat = tmps; doFormatConfig = true; } if (doFormatConfig) this->UpdateFormatConfig(); bool doStreamConfig = false; if (r.Get("Streams",tmps)) { fStreamList = tmps; doStreamConfig = true; } if (doStreamConfig) this->UpdateStreamConfig(); // DDS options bool doDDSConfig = false; if (r.Get("DDSServer", tmps)) { fServer = tmps; doDDSConfig = true; } if (r.Get("DDSPort", tmpi)) { fPort = tmpi; doDDSConfig = true; } if (r.Get("DDSTimeOut",tmpi)) { fTimeOut = tmpi; doDDSConfig = true; } if (r.Get("DDSClientType",tmps)) { fClientType = DDS::GetClientType(tmps); doDDSConfig = true; } if (r.Get("DDSClientName",tmps)) { fClientName = tmps; doDDSConfig = true; } if (r.Get("DDSDataSource",tmps)){fDataSource = DDS::GetDataSourceCode(tmps); doDDSConfig = true;} if (r.Get("DDSKeepUpMode",tmps)) { fKeepUpMode = DDS::GetKeepUpCode(tmps); doDDSConfig = true; } if (r.Get("DDSMaxSyncDelay",tmpi)){fMaxSyncDelay = tmpi; doDDSConfig = true;} if (r.Get("DDSOffLine",tmpb)) {fOffLine = tmpb; doDDSConfig = true;} if (r.Get("DDSMaxRetry",tmpi)) {fMaxRetry = tmpi; doDDSConfig = true; } if (r.Get("DDSRetryDelay",tmpi)) {fRetryDelay = tmpi; doDDSConfig = true; } if (doDDSConfig) this->UpdateDDSConfig(); }

const Registry & IoInputModule::DefaultConfig (   )  const [virtual]

Get the default configuration registry. This should normally be overridden. One useful idiom is to implement it like: const Registry& MyModule::DefaultConfig() const { static Registry cfg; // never is destroyed if (cfg.Size()) return cfg; // already filled it // set defaults: cfg.Set("TheAnswer",42); cfg.Set("Units","unknown"); return cfg; } Reimplemented from JobCModule. Definition at line 121 of file IoInputModule.cxx. References gSystem(), Registry::LockValues(), MSG, Registry::Set(), and Registry::UnLockValues(). Referenced by IoInputModuleValidate::IoInputModuleValidate(). { //====================================================================== // Get the default configuration for this module //====================================================================== static Registry r; r.SetName("INPUT.config"); r.UnLockValues(); MSG("Io",Msg::kDebug) << "Loading default config\n"; // Stream config r.Set("Format" ,"input"); r.Set("Streams","DaqMonitor,DaqSnarl,LightInjection"); // r.Set("Streams","DaqMonitor,DaqSnarl,LightInjection,DcsMonitor,DcsAlarm"); // DDS config r.Set("DDSServer", "daqdds.minos-soudan.org"); r.Set("DDSPort", DDS::kPort); r.Set("DDSTimeOut", 120); r.Set("DDSDataSource","Daq"); r.Set("DDSKeepUpMode", "FileKeepUp"); r.Set("DDSMaxSyncDelay",15); r.Set("DDSOffLine",false); r.Set("DDSMaxRetry",0); r.Set("DDSRetryDelay",1); r.Set("DDSClientType","Unknown"); r.Set("DDSClientName",""); #ifdef SITE_HAS_SAM // SAM config r.Set("Station","minos"); r.Set("SnapShotVers",0); r.Set("WorkGroupName","minos"); r.Set("ApplicationName","loon"); r.Set("ApplicationVers","dev"); r.Set("MaxNumberOfFiles",0); r.Set("StartNewProject",1); // Create default project name // Get $USER const char* username = gSystem->Getenv("USER"); if (!username) username = "unknown"; r.Set("ProjectName",username); #endif r.LockValues(); return r; }

void IoInputModule::DefineStream (  const char *  stream, const char *  tree )  [virtual]

Reimplemented from JobCInputModule. Definition at line 945 of file IoInputModule.cxx. References IoDataStreamItr::DefineStream(), fDataStreamItr, and fStreamDefMap. { //====================================================================== // Define stream to serve specified tree //====================================================================== // Insert the definition into the map fStreamDefMap[stream] = tree; // Pass the info on to the data stream if ( fDataStreamItr ) { fDataStreamItr->DefineStream(stream, tree); } }

void IoInputModule::EndJob (   )  [virtual]

Implement for notification of end of job Reimplemented from JobCModule. Definition at line 110 of file IoInputModule.cxx. References fStopwatch, and MSG. { fStopwatch.Stop(); MSG("Io",Msg::kDebug) << "IoInputModule::EndJob, Time(sec), Real " << fStopwatch.RealTime() << ", CPU " << fStopwatch.CpuTime() << endl; }

JobCResult IoInputModule::Get (  MomNavigator *   )  [inline, private, virtual]

Implement if your module needs to read data from some external source and fill mom Reimplemented from JobCInputModule. Definition at line 105 of file IoInputModule.h. { return JobCResult::kAOK; }

JobCResult IoInputModule::Get (   )  [virtual]

Reimplemented from JobCInputModule. Definition at line 219 of file IoInputModule.cxx. References JobCResult::EndOfInputStream(), fDataStreamItr, fFormat, fStatus, fStopwatch, JobCInputModule::GetMom(), IoDataStreamItr::LoadRecords(), MSG, OpenStreamItr(), ReadHeader(), JobCResult::SetEndFile(), JobCResult::SetEndOfInputStream(), JobCResult::SetEndRun(), and UtilString::ToUpper(). Referenced by GoTo(), Next(), Prev(), and IoInputModuleValidate::TestGet(). { //====================================================================== // Load the data records at the current position in the input stream //====================================================================== if ( fDataStreamItr==0 ) { if ( this->OpenStreamItr()==0 ) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->Get(); } MSG("Io",Msg::kVerbose) << "IoInputModule::Get " << endl; fStopwatch.Start(false); MomNavigator* mom = this->GetMom(); assert(mom); mom -> Clear(); // Moving on so clear contents of Mom int nrecord = fDataStreamItr->LoadRecords(mom); bool isDDS = (UtilString::ToUpper(fFormat) == "DDS"); // special treatment required because dds doesn't separate advance from load if ( isDDS && !nrecord ) fStatus.SetEndOfInputStream(); this->ReadHeader(); // sets beginrun/endrun beginfile/endfile fStatus bits if ( fStatus.EndOfInputStream() ) { fStatus.SetEndFile(); fStatus.SetEndRun(); } MSG("Io",Msg::kVerbose) << "IoInputModule::Get returning status " << fStatus << endl; fStopwatch.Stop(); return fStatus; }

const char * IoInputModule::GetCurrentFile (  const char *  streamname = "*"  )  const [virtual]

Return the currently opened input file. This should not be overridden. Use this in BeginFile(). Reimplemented from JobCInputModule. Definition at line 1068 of file IoInputModule.cxx. References done(), fCurrentFileMap, fDataStreamItr, IoDataStreamItr::GetCurrentFile(), mapStrStrItr_t, and MSG. { MSG("Io",Msg::kDebug) << "IoInputModule::GetCurrentFile()" << endl; mapStrStrItr_t it, done=fCurrentFileMap.end(); std::string strmstring = streamname; for (it = fCurrentFileMap.begin(); it !=done; ++it) { MSG("Io",Msg::kVerbose) << "stream: " << setw(16) << it->first << " file: " << it->second <<endl; } for (it = fCurrentFileMap.begin(); it!=done; ++it) { if ( strmstring == it->first ) return it->second.c_str(); } // Sue's original approach if (!fDataStreamItr) return 0; return fDataStreamItr->GetCurrentFile(streamname); }

Int_t IoInputModule::GetCurrentRun (   )  const [inline, virtual]

Return the current run #. This should not be overridden. Use this in BeginRun(). Reimplemented from JobCInputModule. Definition at line 87 of file IoInputModule.h. { return fCurrentRun; };

Int_t IoInputModule::GetCurrentSnarl (   )  const [inline, virtual]

Reimplemented from JobCInputModule. Definition at line 91 of file IoInputModule.h. { return fCurrentSnarl; };

const char * IoInputModule::GetLastFile (  const char *  streamname = "*"  )  const [virtual]

Return the previously opened input file. This should not be overridden. Use this in EndFile(). Reimplemented from JobCInputModule. Definition at line 1089 of file IoInputModule.cxx. References done(), fLastFileMap, mapStrStrItr_t, and MSG. { MSG("Io",Msg::kInfo) << "IoInputModule::GetLastFile()" << endl; mapStrStrItr_t it, done = fLastFileMap.end(); std::string strmstring = streamname; for (it = fLastFileMap.begin(); it!=done; ++it) { MSG("Io",Msg::kVerbose) << "stream: " << setw(16) << it->first << " file: " << it->second <<endl; } for (it = fLastFileMap.begin(); it!=done; ++it) { if ( strmstring == it->first ) return it->second.c_str(); } return 0; }

Int_t IoInputModule::GetLastRun (   )  const [inline, virtual]

Return the previous run #. This should not be overridden. Use this in EndRun(). Reimplemented from JobCInputModule. Definition at line 88 of file IoInputModule.h. { return fLastRun; };

JobCResult IoInputModule::GoTo (  const VldContext &  vld  )  [virtual]

Reimplemented from JobCInputModule. Definition at line 440 of file IoInputModule.cxx. References fDataStreamItr, CallDepth::fsDepth, fStatus, Get(), GoTo(), MSG, and OpenStreamItr(). { //====================================================================== // Go to records that match validity context. If vld is not found, will // GoTo record set one beyond requested validity. //====================================================================== CallDepth d; // Set the input status to "all clear" if (d.fsDepth==1) fStatus = gsAllClear; if (fDataStreamItr==0) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->GoTo(vld); } fStatus |= fDataStreamItr -> GoTo(vld); // Load the current event return this->Get(); }

JobCResult IoInputModule::GoTo (  int  run, int  snarl, int  searchDir = 0 )  [virtual]

Reimplemented from JobCInputModule. Definition at line 372 of file IoInputModule.cxx. References JobCResult::BeginOfInputStream(), JobCResult::EndOfInputStream(), fCurrentRun, fCurrentSnarl, fDataStreamItr, fLastRun, CallDepth::fsDepth, fStatus, Get(), MSG, Next(), OpenStreamItr(), Prev(), and run(). Referenced by GoTo(), and IoInputModuleValidate::TestGoToEOF(). { CallDepth d; if (d.fsDepth==1) fStatus = gsAllClear; if (fDataStreamItr==0) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->GoTo(run,snarl,searchDir); } if ( run == fCurrentRun && snarl == fCurrentSnarl ) return this -> Get(); int dir = searchDir; if (dir==0) { if (run>fLastRun) { dir = 1; } else if (run<fLastRun || (run==fLastRun && fCurrentRun < 0)) { dir = -1; } else { if (snarl>fLastSnarl) { dir = 1; } else { dir = -1; } } } // Move position in the stream looking for run/event number while ( 1 ) { if ( dir > 0 ) { this->Next(); if (fCurrentRun>run) { MSG("Io",Msg::kWarning) << "Went to run "<<fCurrentRun<< " without finding run="<<run<<" snarl="<<snarl<<"\n"; return fStatus; } if (fCurrentRun==run && fCurrentSnarl>snarl) { MSG("Io",Msg::kWarning) << "Went to run="<<fCurrentRun<<" snarl="<<fCurrentSnarl<< " without finding run="<<run<<" snarl="<<snarl<<"\n"; return fStatus; } } if ( dir <0 ) { this->Prev(); if (fCurrentRun<run) { MSG("Io",Msg::kWarning) << "Went to run "<<fCurrentRun<< " without finding run="<<run<<" snarl="<<snarl<<"\n"; return fStatus; } if (fCurrentRun==run && fCurrentSnarl<snarl) { MSG("Io",Msg::kWarning) << "Went to run="<<fCurrentRun<<" snarl="<<fCurrentSnarl<< " without finding run="<<run<<" snarl="<<snarl<<"\n"; return fStatus; } } // Check if we're done if (fCurrentRun == run && fCurrentSnarl == snarl) return fStatus; if (dir>0 && fStatus.EndOfInputStream()) return fStatus; if (dir<0 && fStatus.BeginOfInputStream()) return fStatus; } return fStatus; }

JobCResult IoInputModule::GoToFile (  const char *  filename, const char *  streamlist = "*" )  [virtual]

Reimplemented from JobCInputModule. Definition at line 884 of file IoInputModule.cxx. References fDataStreamItr, CallDepth::fsDepth, fStatus, GoToFile(), MSG, and OpenStreamItr(). { //====================================================================== // Move the stream to a named file //====================================================================== CallDepth d; if (d.fsDepth==1) fStatus = gsAllClear; if (fDataStreamItr==0) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->GoToFile(filename,streamlist); } fStatus |= fDataStreamItr -> GoToFile(filename,streamlist); return fStatus; }

JobCResult IoInputModule::GoToFile (  int  i, const char *  streamlist = "*" )  [virtual]

Reimplemented from JobCInputModule. Definition at line 859 of file IoInputModule.cxx. References fDataStreamItr, CallDepth::fsDepth, fStatus, MSG, and OpenStreamItr(). Referenced by GoToFile(), and IoInputModuleValidate::TestGoToFile(). { //====================================================================== // Move the stream to the nth file in the list (n=0 is first) //====================================================================== CallDepth d; if (d.fsDepth==1) fStatus = gsAllClear; if (fDataStreamItr==0) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->GoToFile(n,streamlist); } fStatus |= fDataStreamItr -> GoToFile(n,streamlist); return fStatus; }

void IoInputModule::List (  const char *  streamlist = "*"  )  const [virtual]

Reimplemented from JobCInputModule. Definition at line 468 of file IoInputModule.cxx. References fDataStreamItr, fFileList, fFormat, and MSGSTREAM. Referenced by IoInputModuleValidate::TestAdd(), IoInputModuleValidate::TestGoToEOF(), IoInputModuleValidate::TestGoToFile(), IoInputModuleValidate::TestNext(), IoInputModuleValidate::TestNextFile(), IoInputModuleValidate::TestPrevFile(), IoInputModuleValidate::TestRemove(), and IoInputModuleValidate::TestStreams(). { //====================================================================== // Print list of files loaded //====================================================================== MsgStream& m = MSGSTREAM("Io",Msg::kInfo); m << "IoInputModule using data format " << fFormat << endl; if ( fDataStreamItr ) { fDataStreamItr -> ListFile(std::cout,streamlist); return; } // data stream not yet open, list files from fFileList m << "File Name\tStream List " << endl; m << "=========\t=========== " << endl; std::list<IoFileListItem>::const_iterator itr = fFileList.begin(); std::list<IoFileListItem>::const_iterator itrEnd = fFileList.end(); for ( ; itr != itrEnd; itr++ ) { m << *itr; } }

void IoInputModule::LoadFilesFromCommandLine (   )  [private]

Definition at line 1111 of file IoInputModule.cxx. References AddFile(), fLoadedCommandLineFiles, JobCEnv::GetFileName(), JobCEnv::GetNfile(), and JobCEnv::Instance(). Referenced by BeginJob(). { //====================================================================== // Load the files listed on the program command line //====================================================================== JobCEnv& jce = JobCEnv::Instance(); if (!fLoadedCommandLineFiles) { for (int i=0; i<jce.GetNfile(); ++i) { this->AddFile(jce.GetFileName(i)); } fLoadedCommandLineFiles = true; } }

JobCResult IoInputModule::Next (  int  n = 1  )  [virtual]

Reimplemented from JobCInputModule. Definition at line 257 of file IoInputModule.cxx. References JobCResult::EndOfInputStream(), fDataStreamItr, CallDepth::fsDepth, fStatus, fStopwatch, Get(), JobCInputModule::GetMom(), IoDataStreamItr::Increment(), MSG, NextFile(), and OpenStreamItr(). Referenced by GoTo(), IoInputModuleValidate::TestGet(), and IoInputModuleValidate::TestNext(). { //====================================================================== // Advance the position in the stream n record sets. Load the records // at the last position //====================================================================== CallDepth d; // Keep track of the call depth // Set the input status to "all clear" since advancing if (d.fsDepth==1) fStatus = gsAllClear; MSG("Io",Msg::kVerbose) << "IoInputModule::Next " << n << endl; if ( fDataStreamItr==0 ) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->Next(n); } fStopwatch.Start(false); MomNavigator* mom = this->GetMom(); assert(mom); mom -> Clear(); // Moving on so clear contents of Mom // Advance the position in the input stream until we run out of // records and files int nstep = 0; int ndone = 0; int ntry = 0; while ( ndone < n ) { ntry = n - ndone; nstep = fDataStreamItr->Increment(ntry); if ( nstep < ntry ) { // Reached end of file, load next one fStatus |= this->NextFile(); // If this is the end of the input stream, we're done. if ( fStatus.EndOfInputStream() ) { fStopwatch.Stop(); return this->Get(); } } ndone += nstep; } // Load the current event fStopwatch.Stop(); return this->Get(); }

JobCResult IoInputModule::NextFile (  int  n = 1, const char *  streamlist = "*" )  [virtual]

Reimplemented from JobCInputModule. Definition at line 805 of file IoInputModule.cxx. References fDataStreamItr, CallDepth::fsDepth, fStatus, IoDataStreamItr::GetCurrentFile(), MSG, and OpenStreamItr(). Referenced by Next(), and IoInputModuleValidate::TestNextFile(). { //====================================================================== // Move to the next file in the list (move by n positions) //====================================================================== CallDepth d; if (d.fsDepth==1) fStatus = gsAllClear; if (fDataStreamItr==0) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->NextFile(n,streamlist); } fStatus |= fDataStreamItr -> NextFile(n,streamlist); MSG("Io",Msg::kDebug) << "status is " << fStatus << " current file is " << fDataStreamItr->GetCurrentFile() << endl; return fStatus; }

int IoInputModule::OpenStreamItr (   )  [private]

Definition at line 1473 of file IoInputModule.cxx. References CloseStreamItr(), IoDataStreamFactory::CreateDataStreamItr(), fDataStreamItr, fFormat, fStatus, IoDataStreamItr::GetFormat(), MSG, JobCResult::SetBeginFile(), JobCResult::SetBeginOfInputStream(), JobCResult::SetBeginRun(), JobCResult::SetEndFile(), JobCResult::SetEndOfInputStream(), JobCResult::SetEndRun(), UtilString::ToUpper(), UpdateDDSConfig(), UpdateFileList(), and UpdateStreamConfig(). Referenced by Get(), GoTo(), GoToFile(), Next(), NextFile(), Prev(), and PrevFile(). { //====================================================================== // Open a new stream iterator //====================================================================== if (fDataStreamItr) this->CloseStreamItr(); bool isDDS = (UtilString::ToUpper(fFormat) == "DDS"); std::string src; if ( isDDS ) src = fServer; fDataStreamItr = IoDataStreamFactory::CreateDataStreamItr(src.c_str(), fFormat.c_str(),fPort,fMaxRetry,fRetryDelay, fClientType,fClientName); if (fDataStreamItr == 0) { MSG("Io",Msg::kWarning) << "Failed to open stream '" << src << "'" << " using format '" << fFormat << "'" << endl; fStatus.SetEndRun(); fStatus.SetEndFile(); fStatus.SetEndOfInputStream(); return 0; } fStatus.SetBeginOfInputStream(); fStatus.SetBeginFile(); fStatus.SetBeginRun(); // Configure the file and module this->UpdateStreamConfig(); // this should be called before filelist this->UpdateFileList(); this->UpdateDDSConfig(); fFormat = fDataStreamItr->GetFormat(); MSG("Io",Msg::kDebug) << "Opened stream itr of format " << fDataStreamItr->GetFormat() << endl; return 1; }

JobCResult IoInputModule::Prev (  int  n = 1  )  [virtual]

Reimplemented from JobCInputModule. Definition at line 312 of file IoInputModule.cxx. References JobCResult::BeginOfInputStream(), IoDataStreamItr::Decrement(), JobCResult::EndOfInputStream(), fDataStreamItr, CallDepth::fsDepth, fStatus, fStopwatch, Get(), JobCInputModule::GetMom(), IoDataStreamItr::GoToEOF(), MSG, OpenStreamItr(), and PrevFile(). Referenced by GoTo(), and IoInputModuleValidate::TestPrev(). { //====================================================================== // Back up n positions in the input data stream. Load the records at // the current position //====================================================================== CallDepth d; // Set the input status to "all clear" if (d.fsDepth == 1) fStatus = gsAllClear; MSG("Io",Msg::kVerbose) << "IoInputModule::Prev " << n << endl; if (fDataStreamItr==0) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->Prev(n); } // Back up the the position in the input stream until we run out of // records and files fStopwatch.Start(false); MomNavigator* mom = this->GetMom(); assert(mom); mom -> Clear(); // Moving on so clear contents of Mom int nstep = 0; int ndone = 0; int ntry = 0; while (ndone < n) { ntry = n - ndone; nstep = fDataStreamItr->Decrement(ntry); if (nstep < ntry) { // Reached start of file, load previous file. fStatus |= this->PrevFile(); // If there is no previous file, then we're done. if ( fStatus.BeginOfInputStream() || fStatus.EndOfInputStream() ) { return this->Get(); // may be end if failed to open stream itr fStopwatch.Stop(); } // Move the position to the end of the current file so we can // walk backwards over it fDataStreamItr->GoToEOF(); } ndone += nstep; } // Load the current event fStopwatch.Stop(); return this->Get(); }

JobCResult IoInputModule::PrevFile (  int  n = 1, const char *  streamlist = "*" )  [virtual]

Reimplemented from JobCInputModule. Definition at line 834 of file IoInputModule.cxx. References fDataStreamItr, CallDepth::fsDepth, fStatus, MSG, and OpenStreamItr(). Referenced by Prev(), and IoInputModuleValidate::TestPrevFile(). { //====================================================================== // Move to the previous list in the file (move back by n files) //====================================================================== CallDepth d; if (d.fsDepth==1) fStatus = gsAllClear; if (fDataStreamItr==0) { if (this->OpenStreamItr()==0) { MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl; return fStatus; } return this->PrevFile(n,streamlist); } fStatus |= fDataStreamItr -> PrevFile(n,streamlist); return fStatus; }

int IoInputModule::ReadHeader (   )  [private]

Definition at line 1127 of file IoInputModule.cxx. References MomNavigator::At(), JobCResult::BeginFile(), done(), fCurrentFileMap, fCurrentRun, fCurrentSnarl, fLastFileMap, fLastRun, fLastSnarl, fStatus, Registry::Get(), MomNavigator::GetFragmentArray(), RecRecord::GetHeader(), RecMinos::GetHeader(), JobCInputModule::GetMom(), RecDataHeader::GetRun(), CandHeader::GetRun(), RawDaqHeader::GetRun(), RecPhysicsHeader::GetSnarl(), CandHeader::GetSnarl(), RawDaqSnarlHeader::GetSnarl(), RecRecord::GetTempTags(), RecMinos::GetTempTags(), mapStrStrItr_t, MSG, run(), JobCResult::SetBeginFile(), JobCResult::SetBeginRun(), JobCResult::SetEndFile(), and JobCResult::SetEndRun(). Referenced by Get(). { //====================================================================== // Read temptags to get file name // Read header information to get run/snarl info. // if found Run and Snarl info return 2, // if only Run info return 1, // else return 0 //====================================================================== const MomNavigator* mom = this->GetMom(); assert(mom); // BeginFile/EndFile boundaries may not be in synch across the different // data streams. The definition used here is to set file boundary true if // the file has changed for any of the managed streams. const TObjArray* momarray = mom->GetFragmentArray(); for (int i=0; i < momarray->GetEntriesFast(); ++i) { TObject* obj = momarray->At(i); if (!obj) continue; Registry* temptags = 0; if ( RecMinos* record = dynamic_cast<RecMinos*>(obj) ) { temptags = &(record->GetTempTags()); } else if ( RecRecord* record = dynamic_cast<RecRecord*>(obj) ) { temptags = &(record->GetTempTags()); } if ( ! temptags ) continue; // on a named stream? const char* tagstream = 0; if ( ! temptags->Get("stream",tagstream) ) continue; // stream managed by i/o? std::string streamname(tagstream); const char* tagnewfile = 0; if ( ! temptags->Get("file",tagnewfile) ) continue; std::string lstfilename = fLastFileMap[streamname]; std::string curfilename = fCurrentFileMap[streamname]; std::string newfilename(tagnewfile); if ( newfilename != curfilename ) { std::string starcur = fCurrentFileMap.begin()->first; std::string starlast = fLastFileMap.begin()->first; MSG("Io",Msg::kDebug) << "SetBeginFile on streamname '" << streamname << "'" << endl << " current '" << fCurrentFileMap[streamname] << "'" << endl << " last '" << fLastFileMap[streamname] << "'" << endl << " current['" << starcur << "'] '" << fCurrentFileMap[starcur] << "'" << endl << " last['" << starlast << "'] '" << fLastFileMap[starlast] << "'" << endl << " new '" << newfilename << "' != " << " cur '" << curfilename << "'" << endl << " update \"*\" ? " << (( newfilename != fCurrentFileMap["*"] ) ? "yes":"no") << endl; // update "*" stream first if ( newfilename != fCurrentFileMap["*"] ) { fStatus.SetBeginFile(); //if ( lstfilename != "" ) fStatus.SetEndFile(); if ( fCurrentFileMap["*"] != "" ) fStatus.SetEndFile(); fLastFileMap["*"] = fCurrentFileMap["*"]; fCurrentFileMap["*"] = newfilename; MSG("Io",Msg::kDebug) << "SetBeginFile on '*'" << endl << " current['" << starcur << "'] '" << fCurrentFileMap[starcur] << "'" << endl << " last['" << starlast << "'] '" << fLastFileMap[starlast] << "'" << endl; } // update this named stream fLastFileMap[streamname] = curfilename; fCurrentFileMap[streamname] = newfilename; // if a stream on a file moved on then presumably all the // other streams on the same file have also been exhausted // and are going to move on -- help them along so we don't // have to wait for that stream to be the next record on // that stream is the next VldContext mapStrStrItr_t it, done = fCurrentFileMap.end(); for (it = fCurrentFileMap.begin(); it != done; ++it) { if ( it->second == curfilename ) { std::string altstream = it->first; fLastFileMap[altstream] = curfilename; fCurrentFileMap[altstream] = newfilename; } } } // new != cur filename MSG("Io",Msg::kVerbose) << " stream '" << streamname << "' set fLastFileMap to '" << curfilename << "', fCurrentFileMap to '" << newfilename << "'" << " * '" << fLastFileMap["*"] << "' '" << fCurrentFileMap["*"] << "'" << endl; } // loop over records // update for EOF/EOJ condition (which doesn't come throught this // function) if it isn't a file change // this won't work is the file has just one record // ...the whole procedure is fundamentally flawed -- it shouldn't be // based on what we find in "mom" but rather the stream/file // management classes... if ( ! fStatus.BeginFile() ) fLastFileMap["*"] = fCurrentFileMap["*"]; // BeginRun/EndRun int run = -1; // default and flag value int snarl = -1; // default and flag value for (int i=0; i < momarray->GetEntriesFast(); ++i) { TObject* obj = momarray->At(i); if (!obj) continue; if ( RecMinos* record = dynamic_cast<RecMinos*>(obj) ) { // old style // all DAQ generated records can supply run # const RawDaqHeader* rdh = dynamic_cast<const RawDaqHeader*>(record->GetHeader()); if (rdh) run = rdh->GetRun(); // but only DaqSnarl records can supply snarl # const RawDaqSnarlHeader* rdsh = dynamic_cast<const RawDaqSnarlHeader*>(record->GetHeader()); if (rdsh) snarl = rdsh->GetSnarl(); if (!rdh) { // not a DAQ record, perhaps it's a CandRecord const CandHeader* candhdr = dynamic_cast<const CandHeader*>(record->GetHeader()); if (candhdr) { run = candhdr->GetRun(); snarl = candhdr->GetSnarl(); } } } else if ( RecRecord* record = dynamic_cast<RecRecord*>(obj) ) { // New style const RecPhysicsHeader* rph = dynamic_cast<const RecPhysicsHeader*>(&(record->GetHeader())); if ( rph ) { run = rph->GetRun(); snarl = rph->GetSnarl(); } } // break early only if determined snarl in case it one of those // crazy record sets with a DaqMonitor and a DaqSnarl record. if ( snarl >= 0 ) break; } // set the status flags based on what was extracted fCurrentSnarl = snarl; if ( run < 0 ) { fCurrentRun = -1; return 0; } if ( run != fCurrentRun ) { fStatus.SetBeginRun(); if ( fLastRun >= 0 ) fStatus.SetEndRun(); } fLastRun = fCurrentRun; fCurrentRun = run; if ( snarl >= 0 ) { fLastSnarl = fCurrentSnarl; fCurrentSnarl = snarl; return 2; } return 1; }

void IoInputModule::RemoveFile (  const char *  filename, const char *  streamlist = "*" )  [virtual]

Reimplemented from JobCInputModule. Definition at line 783 of file IoInputModule.cxx. References fDataStreamItr, fFileList, IoFileListItem::GetNumFile(), and IoFileListItem::RemoveFile(). Referenced by IoInputModuleValidate::TestRemove(), and UpdateFileList(). { //====================================================================== // Remove the file "filename" from the list of input data files. //====================================================================== if ( fDataStreamItr ) fDataStreamItr -> RemoveFile(filename,streamlist); std::string f(filename); std::list<IoFileListItem>::iterator itr = fFileList.end(); while ( !fFileList.empty() && itr != fFileList.begin() ) { itr--; IoFileListItem& iofile = *itr; iofile.RemoveFile(filename,streamlist); if ( iofile.GetNumFile() == 0 ) fFileList.erase(itr); } return; }

void IoInputModule::Select (  const char *  stream, const char *  select, bool  isRequired = false )  [virtual]

Reimplemented from JobCInputModule. Definition at line 907 of file IoInputModule.cxx. References fDataStreamItr, fStreamRequiredMap, fStreamSelectionMap, and IoDataStreamItr::Select(). { //====================================================================== // Add/Change the selection cuts for a stream //====================================================================== // Insert the selection into the map fStreamSelectionMap[stream] = select; fStreamRequiredMap[stream] = isRequired; // Pass the info on to the data stream if (fDataStreamItr) { fDataStreamItr->Select(stream, select,isRequired); } }

void IoInputModule::SetMaxFileRepeat (  const char *  stream, int  numRepeat )  [virtual]

Reimplemented from JobCInputModule. Definition at line 1011 of file IoInputModule.cxx. References fDataStreamItr, fStreamMaxRepeatMap, and IoDataStreamItr::SetMaxFileRepeat(). { //====================================================================== // Define maximum number of times to reuse a file in the stream before // loading the next one; for kSequential and kRandom sequence modes //====================================================================== fStreamMaxRepeatMap[stream] = numRepeat; // Pass it on to the data stream if ( fDataStreamItr ) fDataStreamItr->SetMaxFileRepeat(stream,numRepeat); }

void IoInputModule::SetMeanMom (  const char *  stream, double  mean )  [virtual]

Reimplemented from JobCInputModule. Definition at line 1025 of file IoInputModule.cxx. References fDataStreamItr, fStreamMeanMap, and IoDataStreamItr::SetMeanMom(). { //====================================================================== // Define mean number of events to push to mom for this stream // for kSequential and kRandom sequence modes //====================================================================== fStreamMeanMap[stream] = mean; // Pass it on to the data stream if ( fDataStreamItr ) fDataStreamItr->SetMeanMom(stream,mean); }

void IoInputModule::SetPerOwnedDisabled (  const char *  stream, bool  perowneddisabled = true )  [virtual]

Reimplemented from JobCInputModule. Definition at line 926 of file IoInputModule.cxx. References fDataStreamItr, fStreamPerOwnedDisabledMap, and IoDataStreamItr::SetPerOwnedDisabled(). Referenced by UpdateStreamConfig(). { //====================================================================== // Used to disable Persistency Ownership of records for this stream //====================================================================== // Insert the perowneddisabled bool into the map fStreamPerOwnedDisabledMap[stream] = perowneddisabled; // Pass the info on to the data stream if (fDataStreamItr) { fDataStreamItr->SetPerOwnedDisabled(stream, perowneddisabled); } }

void IoInputModule::SetPushRandom (  const char *  stream, bool  setRandom )  [virtual]

Reimplemented from JobCInputModule. Definition at line 1039 of file IoInputModule.cxx. References fDataStreamItr, fStreamPushRandomMap, and IoDataStreamItr::SetPushRandom(). { //====================================================================== // Define whether to push a random or constant number of events to mom // for this stream for kSequential and kRandom sequence modes //====================================================================== fStreamPushRandomMap[stream] = setRandom; // Pass it on to the data stream if ( fDataStreamItr ) fDataStreamItr->SetPushRandom(stream,setRandom); }

void IoInputModule::SetRandomSeed (  int  rSeed  )  [virtual]

Reimplemented from JobCInputModule. Definition at line 1053 of file IoInputModule.cxx. References fDataStreamItr, fRandomSeed, and IoDataStreamItr::SetRandomSeed(). { //====================================================================== // Set the random seed for SetPushRandom(stream,true) case // for kSequential and kRandom sequence modes //====================================================================== fRandomSeed = rSeed; // Pass it on to the data stream if ( fDataStreamItr ) fDataStreamItr->SetRandomSeed(rSeed); }

void IoInputModule::SetSequenceMode (  const char *  stream, Per::ESequenceMode  sequenceMode = Per::kKey )  [virtual]

Reimplemented from JobCInputModule. Definition at line 961 of file IoInputModule.cxx. References fDataStreamItr, fStreamSeqModeMap, and IoDataStreamItr::SetSequenceMode(). { //====================================================================== // Define stream sequence mode //====================================================================== // Insert the sequence mode into the map fStreamSeqModeMap[stream] = sequenceMode; // Pass the info on to the data stream if ( fDataStreamItr ) { fDataStreamItr->SetSequenceMode(stream, sequenceMode); } }

void IoInputModule::SetTestMode (  const char *  stream, bool  testMode )  [virtual]

Reimplemented from JobCInputModule. Definition at line 978 of file IoInputModule.cxx. References fDataStreamItr, fStreamTestModeMap, and IoDataStreamItr::SetTestMode(). { //====================================================================== // Define stream test mode //====================================================================== // Insert the test mode into the map fStreamTestModeMap[stream] = testMode; // Pass the info on to the data stream if ( fDataStreamItr ) { fDataStreamItr->SetTestMode(stream, testMode); } }

void IoInputModule::SetWindow (  const char *  stream, double  lower, double  upper )  [virtual]

Reimplemented from JobCInputModule. Definition at line 995 of file IoInputModule.cxx. References fDataStreamItr, fStreamWindowMap, and IoDataStreamItr::SetWindow(). { //====================================================================== // Define stream window if kWindow sequence mode is used //====================================================================== fStreamWindowMap[stream] = std::pair<double,double>(lower,upper); // Pass the info on to the data stream if ( fDataStreamItr ) { fDataStreamItr->SetWindow(stream, lower, upper); } }

void IoInputModule::UpdateDDSConfig (   )  [private]

Definition at line 1306 of file IoInputModule.cxx. References CloseStreamItr(), fClientName, fClientType, fDataSource, fDataStreamItr, fKeepUpMode, fMaxSyncDelay, fOffLine, fPort, fServer, fTimeOut, IoDDSStreamItr::GetClientName(), IoDDSStreamItr::GetClientType(), IoDDSStreamItr::GetPort(), IoDataStreamItr::GetSourceName(), and IoDDSStreamItr::SetTimeOut(). Referenced by Config(), and OpenStreamItr(). { //====================================================================== // Update the dispatcher configuration //====================================================================== if ( fDataStreamItr == 0 ) return; IoDDSStreamItr* ddsItr = dynamic_cast<IoDDSStreamItr*>(fDataStreamItr); if ( ! ddsItr ) return; ddsItr->SetTimeOut(fTimeOut); // Need to reinitialize dispatcher if server hostname, port, clienttype // or clientname have changed bool reinit = (fServer != ddsItr->GetSourceName() || fPort != ddsItr->GetPort() || fClientType != ddsItr->GetClientType() || fClientName != ddsItr->GetClientName() ); if ( reinit ) { this -> CloseStreamItr(); // wait for next action to reopen } else { ddsItr -> SetKeepUpMode(fKeepUpMode); ddsItr -> SetMaxSyncDelay(fMaxSyncDelay); ddsItr -> SetDataSource(fDataSource); ddsItr -> SetOffLine(fOffLine); } }

void IoInputModule::UpdateFileList (   )  [private]

Definition at line 1357 of file IoInputModule.cxx. References AddFile(), fDataStreamItr, fFileList, IoFileListItem::GetAt(), IoFileListItem::GetFileStreamMap(), and RemoveFile(). Referenced by OpenStreamItr(). { //====================================================================== // Update the file list //====================================================================== if ( fDataStreamItr == 0 ) return; // Fresh start. This should typically only be called when data stream // iterator is newly opened (i.e. when the format changes) fDataStreamItr -> RemoveFile("*"); std::list<IoFileListItem>::iterator itr = fFileList.begin(); for ( ; itr != fFileList.end(); itr++ ) { IoFileListItem& iofile = *itr; const IoFileListItem::FileStreamMap& filemap = iofile.GetFileStreamMap(); int at = iofile.GetAt(); if ( at < 0 ) { IoFileListItem::FileStreamMapConstItr itr = filemap.begin(); for ( ; itr != filemap.end(); itr++ ) { std::string filename = itr -> first; std::string streamlist = itr -> second; fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str()); } } else { // Apply files in reverse to have first file in list inserted at pos At IoFileListItem::FileStreamMap::const_reverse_iterator itr = filemap.rbegin(); for ( ; itr != filemap.rend(); itr++ ) { std::string filename = itr -> first; std::string streamlist = itr -> second; fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str()); } } } return; }

void IoInputModule::UpdateFormatConfig (   )  [private]

Definition at line 1339 of file IoInputModule.cxx. References CloseStreamItr(), fDataStreamItr, fFormat, and IoDataStreamItr::GetFormat(). Referenced by Config(). { //====================================================================== // Update the stream itr to match requested format //====================================================================== if ( fDataStreamItr == 0 ) return; bool reopen = ( fFormat != fDataStreamItr->GetFormat() ); if ( reopen ) { this -> CloseStreamItr(); // wait for next action to reopen } return; }

void IoInputModule::UpdateStreamConfig (   )  [private]

Definition at line 1401 of file IoInputModule.cxx. References IoDataStreamItr::DefineStream(), fDataStreamItr, fRandomSeed, fStreamDefMap, fStreamList, fStreamMaxRepeatMap, fStreamMeanMap, fStreamPerOwnedDisabledMap, fStreamPushRandomMap, fStreamRequiredMap, fStreamSelectionMap, fStreamSeqModeMap, fStreamTestModeMap, fStreamWindowMap, mapStrStrItr_t, IoDataStreamItr::Select(), IoDataStreamItr::SetMaxFileRepeat(), IoDataStreamItr::SetMeanMom(), SetPerOwnedDisabled(), IoDataStreamItr::SetPushRandom(), IoDataStreamItr::SetRandomSeed(), IoDataStreamItr::SetSequenceMode(), IoDataStreamItr::SetTestMode(), IoDataStreamItr::SetWindow(), and IoDataStreamItr::Streams(). Referenced by Config(), and OpenStreamItr(). { //====================================================================== // Set the stream and selection cuts for the open streams //====================================================================== if (fDataStreamItr==0) return; // Define streams mapStrStrItr_t itr; for ( itr = fStreamDefMap.begin(); itr != fStreamDefMap.end(); itr++ ) { fDataStreamItr->DefineStream((itr->first).c_str(), // Stream (itr->second).c_str()); // Definition } // Set the streams to be activated fDataStreamItr->Streams(fStreamList.c_str()); // Set the selection cuts for each stream for (itr=fStreamSelectionMap.begin();itr!=fStreamSelectionMap.end();++itr) { std::map<std::string,bool>::const_iterator reqitr = fStreamRequiredMap.find(itr->first); bool isrequired = false; if ( reqitr != fStreamRequiredMap.end() ) isrequired = reqitr -> second; fDataStreamItr->Select((itr->first).c_str(), // Stream (itr->second).c_str(), // Selection isrequired); // IsRequired } // Set the test mode for each stream std::map<std::string,bool>::const_iterator testitr; for ( testitr=fStreamTestModeMap.begin(); testitr !=fStreamTestModeMap.end(); ++testitr ) { fDataStreamItr->SetTestMode((testitr->first).c_str(), // Stream testitr->second); // TestMode } // Disable per owned for each stream requested std::map<std::string,bool>::const_iterator disableitr; for ( disableitr = fStreamPerOwnedDisabledMap.begin(); disableitr != fStreamPerOwnedDisabledMap.end(); ++disableitr ) { fDataStreamItr -> SetPerOwnedDisabled((disableitr->first).c_str(), disableitr->second); } // Set the sequence mode for each stream bool setRandom = false; std::map<std::string,Per::ESequenceMode>::const_iterator seqitr; for ( seqitr=fStreamSeqModeMap.begin(); seqitr!=fStreamSeqModeMap.end(); ++seqitr ) { fDataStreamItr->SetSequenceMode((seqitr->first).c_str(), // Stream seqitr->second); // Sequence Mode pair<double,double> window = fStreamWindowMap[seqitr->first]; fDataStreamItr->SetWindow((seqitr->first).c_str(), // Stream window.first,window.second); if (seqitr->second == Per::kSequential || seqitr->second == Per::kRandom ) { if (!setRandom) { setRandom = true; fDataStreamItr->SetRandomSeed(fRandomSeed); } int repeat = fStreamMaxRepeatMap[seqitr->first]; fDataStreamItr->SetMaxFileRepeat( (seqitr->first).c_str(), repeat ); double mean = fStreamMeanMap[seqitr->first]; fDataStreamItr->SetMeanMom( (seqitr->first).c_str(), mean ); bool pushRand = fStreamPushRandomMap[seqitr->first]; fDataStreamItr->SetPushRandom( (seqitr->first).c_str(), pushRand ); } // end if kSeq or kRand } }

---

Member Data Documentation

string IoInputModule::fClientName [private]

Definition at line 138 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

DDS::EClientType IoInputModule::fClientType [private]

Definition at line 137 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

std::map<std::string,std::string> IoInputModule::fCurrentFileMap [private]

Definition at line 146 of file IoInputModule.h. Referenced by GetCurrentFile(), and ReadHeader().

int IoInputModule::fCurrentRun [private]

Definition at line 144 of file IoInputModule.h. Referenced by GoTo(), and ReadHeader().

int IoInputModule::fCurrentSnarl [private]

Definition at line 145 of file IoInputModule.h. Referenced by GoTo(), and ReadHeader().

int IoInputModule::fDataSource [private]

Definition at line 131 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

IoDataStreamItr* IoInputModule::fDataStreamItr [private]

Definition at line 117 of file IoInputModule.h. Referenced by AddFile(), CloseStreamItr(), DefineStream(), Get(), GetCurrentFile(), GoTo(), GoToFile(), List(), Next(), NextFile(), OpenStreamItr(), Prev(), PrevFile(), RemoveFile(), Select(), SetMaxFileRepeat(), SetMeanMom(), SetPerOwnedDisabled(), SetPushRandom(), SetRandomSeed(), SetSequenceMode(), SetTestMode(), SetWindow(), UpdateDDSConfig(), UpdateFileList(), UpdateFormatConfig(), UpdateStreamConfig(), and ~IoInputModule().

std::list<IoFileListItem> IoInputModule::fFileList [private]

Definition at line 116 of file IoInputModule.h. Referenced by AddFile(), List(), RemoveFile(), and UpdateFileList().

std::string IoInputModule::fFormat [private]

Definition at line 119 of file IoInputModule.h. Referenced by Config(), Get(), List(), OpenStreamItr(), and UpdateFormatConfig().

int IoInputModule::fKeepUpMode [private]

Definition at line 132 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

std::string IoInputModule::fLastBeginFile [private]

Definition at line 148 of file IoInputModule.h.

std::string IoInputModule::fLastEndFile [private]

Definition at line 149 of file IoInputModule.h.

std::map<std::string,std::string> IoInputModule::fLastFileMap [private]

Definition at line 147 of file IoInputModule.h. Referenced by GetLastFile(), and ReadHeader().

int IoInputModule::fLastRun [private]

Definition at line 142 of file IoInputModule.h. Referenced by GoTo(), and ReadHeader().

int IoInputModule::fLastSnarl [private]

Definition at line 143 of file IoInputModule.h. Referenced by ReadHeader().

bool IoInputModule::fLoadedCommandLineFiles [private]

Definition at line 150 of file IoInputModule.h. Referenced by LoadFilesFromCommandLine().

int IoInputModule::fMaxRetry [private]

Definition at line 135 of file IoInputModule.h. Referenced by Config().

int IoInputModule::fMaxSyncDelay [private]

Definition at line 133 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

bool IoInputModule::fOffLine [private]

Definition at line 134 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

int IoInputModule::fPort [private]

Definition at line 129 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

int IoInputModule::fRandomSeed [private]

Definition at line 125 of file IoInputModule.h. Referenced by SetRandomSeed(), and UpdateStreamConfig().

int IoInputModule::fRetryDelay [private]

Definition at line 136 of file IoInputModule.h. Referenced by Config().

std::string IoInputModule::fServer [private]

Definition at line 128 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

JobCResult IoInputModule::fStatus [private]

Definition at line 141 of file IoInputModule.h. Referenced by CloseStreamItr(), Get(), GoTo(), GoToFile(), Next(), NextFile(), OpenStreamItr(), Prev(), PrevFile(), and ReadHeader().

TStopwatch IoInputModule::fStopwatch [private]

Definition at line 151 of file IoInputModule.h. Referenced by EndJob(), Get(), IoInputModule(), Next(), and Prev().

std::map<std::string,std::string> IoInputModule::fStreamDefMap [private]

Definition at line 110 of file IoInputModule.h. Referenced by DefineStream(), and UpdateStreamConfig().

std::string IoInputModule::fStreamList [private]

Definition at line 120 of file IoInputModule.h. Referenced by Config(), and UpdateStreamConfig().

std::map<std::string,int> IoInputModule::fStreamMaxRepeatMap [private]

Definition at line 122 of file IoInputModule.h. Referenced by SetMaxFileRepeat(), and UpdateStreamConfig().

std::map<std::string,double> IoInputModule::fStreamMeanMap [private]

Definition at line 123 of file IoInputModule.h. Referenced by SetMeanMom(), and UpdateStreamConfig().

std::map<std::string,bool> IoInputModule::fStreamPerOwnedDisabledMap [private]

Definition at line 112 of file IoInputModule.h. Referenced by SetPerOwnedDisabled(), and UpdateStreamConfig().

std::map<std::string,bool> IoInputModule::fStreamPushRandomMap [private]

Definition at line 124 of file IoInputModule.h. Referenced by SetPushRandom(), and UpdateStreamConfig().

std::map<std::string,bool> IoInputModule::fStreamRequiredMap [private]

Definition at line 109 of file IoInputModule.h. Referenced by Select(), and UpdateStreamConfig().

std::map<std::string,std::string> IoInputModule::fStreamSelectionMap [private]

Definition at line 108 of file IoInputModule.h. Referenced by Select(), and UpdateStreamConfig().

std::map<std::string,Per::ESequenceMode> IoInputModule::fStreamSeqModeMap [private]

Definition at line 111 of file IoInputModule.h. Referenced by SetSequenceMode(), and UpdateStreamConfig().

std::map<std::string,bool> IoInputModule::fStreamTestModeMap [private]

Definition at line 113 of file IoInputModule.h. Referenced by SetTestMode(), and UpdateStreamConfig().

std::map<std::string,std::pair<double,double> > IoInputModule::fStreamWindowMap [private]

Definition at line 114 of file IoInputModule.h. Referenced by SetWindow(), and UpdateStreamConfig().

int IoInputModule::fTimeOut [private]

Definition at line 130 of file IoInputModule.h. Referenced by Config(), and UpdateDDSConfig().

---

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

• IoInputModule.h
• IoInputModule.cxx

---