evf::EvFDaqDirector Class Reference

#include <EvFDaqDirector.h>

Public Member Functions
std::string &	baseDir ()
std::string &	buBaseDir ()
std::string &	buBaseOpenDir ()
	EvFDaqDirector (const edm::ParameterSet &pset, edm::ActivityRegistry &reg)
std::string	findCurrentRunDir ()
unsigned int	findHighestRun ()
std::string	findHighestRunDir ()
std::string	findHighestRunDirStem ()
std::string &	fuBaseDir ()
std::string	getEoLSFilePathOnBU (const unsigned int ls) const
std::string	getEoLSFilePathOnFU (const unsigned int ls) const
std::string	getEoRFilePath () const
std::string	getInitFilePath (std::string const &stream) const
std::string	getJumpFilePath () const
unsigned int	getJumpIndex () const
unsigned int	getJumpLS () const
std::string	getMergedDatFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenDatFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenRawFilePath (const unsigned int ls, const unsigned int index) const
std::string	getOutputJsonFilePath (const unsigned int ls, std::string const &stream) const
std::string	getPathForFU () const
std::string	getRawFilePath (const unsigned int ls, const unsigned int index) const
unsigned int	getRunNumber () const
bool	getTestModeNoBuilderUnit ()
FILE *	maybeCreateAndLockFileHeadForStream (unsigned int ls, std::string &stream)
std::string &	monitorBaseDir ()
void	postEndRun (edm::Run const &run, edm::EventSetup const &es)
void	preBeginRun (edm::RunID const &id, edm::Timestamp const &ts)
int	readBuLock ()
void	removeFile (unsigned int ls, unsigned int index)
void	removeFile (std::string)
std::string &	smBaseDir ()
void	tryInitializeFuLockFile ()
void	unlockAndCloseMergeStream ()
void	updateBuLock (unsigned int ls)
bool	updateFuLock (unsigned int &ls, std::string &nextFile, bool &eorSeen)
void	writeDiskAndThrottleStat (double, int, int)
void	writeLsStatisticsBU (unsigned int, unsigned int, unsigned long long, long long)
void	writeLsStatisticsFU (unsigned int ls, unsigned int events, timeval completion_time)
	~EvFDaqDirector ()

Private Member Functions
bool	bulock ()
bool	bumpFile (unsigned int &ls, unsigned int &index, std::string &nextFile)
bool	createOutputDirectory ()
std::string	eolsFileName (const unsigned int ls) const
std::string	eorFileName () const
bool	findHighestActiveLS (unsigned int &startingLS) const
bool	fulock ()
std::string	initFileName (std::string const &stream) const
std::string	inputFileNameStem (const unsigned int ls, const unsigned int index) const
std::string	mergedFileNameStem (const unsigned int ls, std::string const &stream) const
bool	mkFuRunDir ()
void	openFULockfileStream (std::string &fuLockFilePath, bool create)
std::string	outputFileNameStem (const unsigned int ls, std::string const &stream) const

Private Attributes
std::string	base_dir_
std::string	bu_base_dir_
struct flock	bu_r_flk
struct flock	bu_r_fulk
FILE *	bu_r_lock_stream
int	bu_readlock_fd_
std::string	bu_run_dir_
std::string	bu_run_open_dir_
FILE *	bu_t_monitor_stream
struct flock	bu_w_flk
struct flock	bu_w_fulk
FILE *	bu_w_lock_stream
FILE *	bu_w_monitor_stream
int	bu_writelock_fd_
int	data_readwrite_fd_
struct flock	data_rw_flk
struct flock	data_rw_fulk
FILE *	data_rw_stream
bool	directorBu_
DirManager	dirManager_
int	fu_readwritelock_fd_
struct flock	fu_rw_flk
struct flock	fu_rw_fulk
FILE *	fu_rw_lock_stream
std::string	hostname_
unsigned int	jumpIndex_
unsigned int	jumpLS_
std::string	monitor_base_dir_
unsigned long	previousFileSize_
unsigned int	run_
std::string	run_dir_
std::string	run_string_
std::string	sm_base_dir_
bool	testModeNoBuilderUnit_

Detailed Description

Definition at line 25 of file EvFDaqDirector.h.

Constructor & Destructor Documentation

evf::EvFDaqDirector::EvFDaqDirector ( const edm::ParameterSet &  pset, edm::ActivityRegistry &  reg  )

explicit

Definition at line 27 of file EvFDaqDirector.cc.

References base_dir_, bu_base_dir_, bu_run_dir_, bu_run_open_dir_, bu_t_monitor_stream, bu_w_lock_stream, bu_w_monitor_stream, bu_writelock_fd_, createOutputDirectory(), directorBu_, edm::hlt::Exception, lut2db_cfg::filename, fu_readwritelock_fd_, fu_rw_lock_stream, hostname_, monitor_base_dir_, openFULockfileStream(), postEndRun(), preBeginRun(), run_, run_dir_, run_string_, AlCaHLTBitMon_QueryRunRegistry::string, tryInitializeFuLockFile(), edm::ActivityRegistry::watchPostEndRun(), and edm::ActivityRegistry::watchPreBeginRun().

                                           :
    testModeNoBuilderUnit_(
               pset.getUntrackedParameter<bool> ("testModeNoBuilderUnit",
                                 false)
               ),
    base_dir_(
          pset.getUntrackedParameter<std::string> ("baseDir", "/data")
          ),
    bu_base_dir_(
         pset.getUntrackedParameter<std::string> ("buBaseDir", "/data")
         ),
    sm_base_dir_(
         pset.getUntrackedParameter<std::string> ("smBaseDir", "/sm")
         ),
    monitor_base_dir_(
              pset.getUntrackedParameter<std::string> ("monBaseDir",
                                   "MON")
              ),
    directorBu_(
        pset.getUntrackedParameter<bool> ("directorIsBu", false)
        ),
    run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
    hostname_(""),
    bu_readlock_fd_(-1),
    bu_writelock_fd_(-1),
    fu_readwritelock_fd_(-1),
    data_readwrite_fd_(-1),

    bu_w_lock_stream(0),
    bu_r_lock_stream(0),
    fu_rw_lock_stream(0),
    bu_w_monitor_stream(0),
    bu_t_monitor_stream(0),
    data_rw_stream(0),

    dirManager_(base_dir_), 

    previousFileSize_(0), 
    jumpLS_(0),
    jumpIndex_(0),

    bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
    bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
    bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
    bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
    fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
    fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
    data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
    data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
  {
    reg.watchPreBeginRun(this, &EvFDaqDirector::preBeginRun);
    reg.watchPostEndRun(this, &EvFDaqDirector::postEndRun);

    std::stringstream ss;
    ss << "run" << std::setfill('0') << std::setw(6) << run_;
    run_string_ = ss.str();
    run_dir_ = base_dir_+"/"+run_string_;

    //save hostname for later 
    char hostname[33];
    gethostname(hostname,32);
    hostname_ = hostname;
    // check if base dir exists or create it accordingly
    int retval = mkdir(base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error checking for base dir "
                      << base_dir_ << " mkdir error:" << strerror(errno) << "\n";
    }

    //bu_run_dir: for FU, for which the base dir is local and the BU is remote, it is expected to be there
    //for BU, it is created at this point


    if (directorBu_) 
      {
    bu_run_dir_ = bu_base_dir_ + "/" + run_string_;
    std::string bulockfile = bu_run_dir_ + "/bu.lock";
    std::string fulockfile = bu_run_dir_ + "/fu.lock";

    //make or find bu run dir
    retval = mkdir(bu_run_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector")
        << " Error creating bu run dir " << bu_run_dir_
        << " mkdir error:" << strerror(errno) << "\n";
    }
    bu_run_open_dir_ = bu_run_dir_ + "/open";
    retval = mkdir(bu_run_open_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error creating bu run open dir "
                          << bu_run_open_dir_ << " mkdir error:" << strerror(errno)
                          << "\n";
    }
        
    //make or find monitor base dir
    //@@EM make sure this is still needed

    stringstream ost;
    ost << bu_run_dir_ << "/" << monitor_base_dir_;
    monitor_base_dir_ = ost.str() + "_OLD";
    retval = mkdir(monitor_base_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector")
        << " Error creating monitor dir " << monitor_base_dir_
        << " mkdir error:" << strerror(errno) << "\n";
    }
        
    // the BU director does not need to know about the fu lock
    bu_writelock_fd_ = open(bulockfile.c_str(),
                O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
    if (bu_writelock_fd_ == -1)
      edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock "
                        << strerror(errno);
    else
      edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock "
                     << bu_writelock_fd_;
    bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
    if (bu_w_lock_stream == 0)
      edm::LogWarning("EvFDaqDirector")<< "Error creating write lock stream " << strerror(errno);
    std::string filename = monitor_base_dir_ + "/bumonitor.txt";
    bu_w_monitor_stream = fopen(filename.c_str(), "w+");
    filename = monitor_base_dir_ + "/diskmonitor.txt";
    bu_t_monitor_stream = fopen(filename.c_str(), "w+");
    if (bu_t_monitor_stream == 0)
      edm::LogWarning("EvFDaqDirector") << "Error creating bu write lock stream " << strerror(errno);
        
    // BU INITIALIZES LOCK FILE
    // FU LOCK FILE OPEN
    openFULockfileStream(fulockfile, true);
    tryInitializeFuLockFile();
    fflush(fu_rw_lock_stream);
    close(fu_readwritelock_fd_);
    createOutputDirectory(); // this should act not on the bu base dir but on the output disk
      } 
    else 
      {
    // for FU, check if bu base dir exists 
    
    retval = mkdir(bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error checking for bu base dir "
                          << base_dir_ << " mkdir error:" << strerror(errno) << "\n";
    }

    bu_run_dir_ = bu_base_dir_ + "/" + run_string_;
    std::string fulockfile = bu_run_dir_ + "/fu.lock";
    openFULockfileStream(fulockfile, false);
      }


  }

evf::EvFDaqDirector::~EvFDaqDirector ( )

inline

Definition at line 30 of file EvFDaqDirector.h.

{}

Member Function Documentation

std::string& evf::EvFDaqDirector::baseDir ( )

inline

Definition at line 33 of file EvFDaqDirector.h.

References base_dir_.

{return base_dir_;}

std::string& evf::EvFDaqDirector::buBaseDir ( )

inline

Definition at line 36 of file EvFDaqDirector.h.

References bu_run_dir_.

{return bu_run_dir_;}

std::string& evf::EvFDaqDirector::buBaseOpenDir ( )

inline

Definition at line 37 of file EvFDaqDirector.h.

References bu_run_open_dir_.

{return bu_run_open_dir_;}

bool evf::EvFDaqDirector::bulock ( )

private

Definition at line 443 of file EvFDaqDirector.cc.

References bu_base_dir_, bu_readlock_fd_, bu_writelock_fd_, gather_cfg::cout, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by readBuLock().

                              {
    struct stat buf;
    std::string lockfile = bu_base_dir_;
    lockfile += "/bu.lock";
    bool retval = (stat(lockfile.c_str(), &buf) == 0);
    if (!retval) {
      close(bu_readlock_fd_);
      close(bu_writelock_fd_);
    }
    std::cout << "stat of lockfile returned " << retval << std::endl;
    return retval;
  }

bool evf::EvFDaqDirector::bumpFile ( unsigned int &  ls, unsigned int &  index, std::string &  nextFile  )

private

Definition at line 505 of file EvFDaqDirector.cc.

References evf::FastMonitoringService::accummulateFileSize(), newFWLiteAna::found, getEoLSFilePathOnBU(), getRawFilePath(), getHLTprescales::index, python.rootplot.utilities::ls(), previousFileSize_, and testModeNoBuilderUnit_.

Referenced by updateFuLock().

                                                                                          {

    if (previousFileSize_ != 0) {
      FastMonitoringService *mss = (FastMonitoringService *) (edm::Service<
                                  evf::MicroStateService>().operator->());
      mss->accummulateFileSize(previousFileSize_);
      previousFileSize_ = 0;
    }

    struct stat buf;
    std::stringstream ss;
    unsigned int nextIndex = index;
    nextIndex++;

    // 1. Check suggested file
    nextFile = getRawFilePath(ls,index);
    bool found = (stat(nextFile.c_str(), &buf) == 0);
    // if found
    if (found) {
      //grabbedFileSize = buf.st_size;
      previousFileSize_ = buf.st_size;
      return true;
    }
    // 2. No file -> lumi ended? (and how many?)
    else {
      bool eolFound = (stat(getEoLSFilePathOnBU(ls).c_str(), &buf) == 0);
      unsigned int startingLumi = ls;
      while (eolFound) {
    // this lumi ended, check for files
    ++ls;
    nextFile = getRawFilePath(ls,0);
    found = (stat(nextFile.c_str(), &buf) == 0);
    // update highest ls even if there is no file
    // input source can now end its' LS when an EoL jsn file is seen
    if (found) {
      // a new file was found at new lumisection, index 0
      index = 0;
      //grabbedFileSize = buf.st_size;
      previousFileSize_ = buf.st_size;
      
      if (testModeNoBuilderUnit_) {
        // rename ended lumi to + 2
        string sourceEol = getEoLSFilePathOnBU(startingLumi);
        
        string destEol = getEoLSFilePathOnBU(startingLumi+2);
        
        string cpCmd = "cp " + sourceEol + " " + destEol;
        edm::LogInfo("EvFDaqDirector") << " testmode: Running copy cmd = " << cpCmd;
        int rc = system(cpCmd.c_str());
        if (rc != 0) {
          edm::LogError("EvFDaqDirector") << " testmode: COPY EOL FAILED!!!!!: " << cpCmd;
        }
      }
      
      return true;
    }
    eolFound = (stat(getEoLSFilePathOnBU(ls).c_str(), &buf) == 0);
      }
    }
    // no new file found
    return false;
  }

bool evf::EvFDaqDirector::createOutputDirectory ( )

private

Definition at line 216 of file EvFDaqDirector.cc.

References edm::hlt::Exception, run_string_, sm_base_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by EvFDaqDirector().

                                             {
    int retval = mkdir(sm_base_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error creating output dir "
                      << sm_base_dir_ << " mkdir error:" << strerror(errno) << "\n";
      return false;
    }
    std::string mergedRunDir = sm_base_dir_ + "/" + run_string_;
    std::string mergedDataDir = mergedRunDir + "/data";
    std::string mergedMonDir = mergedRunDir + "/mon";
    retval = mkdir(mergedRunDir.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector")
    << " Error creating merged Run dir " << mergedDataDir
    << " mkdir error:" << strerror(errno) << "\n";
      return false;
    }
    retval
      = mkdir(mergedDataDir.c_str(),
          S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector")
    << " Error creating merged data dir " << mergedDataDir
    << " mkdir error:" << strerror(errno) << "\n";
      return false;
    }
    retval = mkdir(mergedMonDir.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector")
    << " Error creating merged mon dir " << mergedMonDir
    << " mkdir error:" << strerror(errno) << "\n";
      return false;
    }
    return true;
  }

std::string evf::EvFDaqDirector::eolsFileName ( const unsigned int  ls ) const

private

Definition at line 661 of file EvFDaqDirector.cc.

Referenced by getEoLSFilePathOnBU(), and getEoLSFilePathOnFU().

                                                                    {
    std::stringstream ss;
    ss << "EoLS_" << std::setfill('0') << std::setw(4) << ls << ".jsn";
    return ss.str();
  }

std::string evf::EvFDaqDirector::eorFileName ( ) const

private

Definition at line 667 of file EvFDaqDirector.cc.

References run_.

Referenced by getEoRFilePath().

                                              {
    std::stringstream ss;
    ss << "EoR_" << std::setfill('0') << std::setw(6) << run_ << ".jsn";
    return ss.str();
  }

std::string evf::EvFDaqDirector::findCurrentRunDir ( )

inline

Definition at line 41 of file EvFDaqDirector.h.

References dirManager_, evf::DirManager::findRunDir(), and run_.

{ return dirManager_.findRunDir(run_);}

bool evf::EvFDaqDirector::findHighestActiveLS ( unsigned int &  startingLS ) const

private

unsigned int evf::EvFDaqDirector::findHighestRun ( )

inline

Definition at line 43 of file EvFDaqDirector.h.

References dirManager_, and evf::DirManager::findHighestRun().

{return dirManager_.findHighestRun();}

std::string evf::EvFDaqDirector::findHighestRunDir ( )

inline

Definition at line 40 of file EvFDaqDirector.h.

References dirManager_, and evf::DirManager::findHighestRunDir().

Referenced by findHighestRunDirStem().

{ return dirManager_.findHighestRunDir();}

std::string evf::EvFDaqDirector::findHighestRunDirStem

(

)

Definition at line 568 of file EvFDaqDirector.cc.

References findHighestRunDir(), and getHLTPrescaleColumns::path.

                                                  {
    boost::filesystem::path highestRunDirPath (findHighestRunDir());
    return highestRunDirPath.filename().string();
  }

std::string& evf::EvFDaqDirector::fuBaseDir ( )

inline

Definition at line 34 of file EvFDaqDirector.h.

References run_dir_.

{return run_dir_;}

bool evf::EvFDaqDirector::fulock ( )

private

Definition at line 456 of file EvFDaqDirector.cc.

References bu_base_dir_, gather_cfg::cout, fu_readwritelock_fd_, and AlCaHLTBitMon_QueryRunRegistry::string.

                              {
    struct stat buf;
    std::string lockfile = bu_base_dir_;
    lockfile += "/fu.lock";
    bool retval = (stat(lockfile.c_str(), &buf) == 0);
    if (!retval) {
      close(fu_readwritelock_fd_);
      close(fu_readwritelock_fd_); // why the second close ?
    }
    std::cout << "stat of lockfile returned " << retval << std::endl;
    return retval;
  }

std::string evf::EvFDaqDirector::getEoLSFilePathOnBU

(

const unsigned int

ls

)

const

Definition at line 277 of file EvFDaqDirector.cc.

References bu_run_dir_, and eolsFileName().

Referenced by bumpFile().

                                                                           {
    return bu_run_dir_ + "/" + eolsFileName(ls);
  }

std::string evf::EvFDaqDirector::getEoLSFilePathOnFU

(

const unsigned int

ls

)

const

Definition at line 281 of file EvFDaqDirector.cc.

References eolsFileName(), and run_dir_.

                                                                           {
    return run_dir_ + "/" + eolsFileName(ls);
  }

std::string evf::EvFDaqDirector::getEoRFilePath

(

)

const

Definition at line 285 of file EvFDaqDirector.cc.

References bu_run_dir_, and eorFileName().

Referenced by updateFuLock().

                                                 {
    return bu_run_dir_ + "/" + eorFileName();
  }

std::string evf::EvFDaqDirector::getInitFilePath

(

std::string const &

stream

)

const

Definition at line 273 of file EvFDaqDirector.cc.

References initFileName(), and run_dir_.

                                                                         {
    return run_dir_ + "/" + initFileName(stream);
  }

std::string evf::EvFDaqDirector::getJumpFilePath ( ) const

inline

Definition at line 68 of file EvFDaqDirector.h.

References bu_run_dir_, inputFileNameStem(), jumpIndex_, and jumpLS_.

{ return bu_run_dir_ + "/" + inputFileNameStem(jumpLS_,jumpIndex_) + ".raw"; }

unsigned int evf::EvFDaqDirector::getJumpIndex ( ) const

inline

Definition at line 67 of file EvFDaqDirector.h.

References jumpIndex_.

{ return jumpIndex_; }

unsigned int evf::EvFDaqDirector::getJumpLS ( ) const

inline

Definition at line 66 of file EvFDaqDirector.h.

References jumpLS_.

{ return jumpLS_; }

std::string evf::EvFDaqDirector::getMergedDatFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 269 of file EvFDaqDirector.cc.

References mergedFileNameStem(), and run_dir_.

Referenced by maybeCreateAndLockFileHeadForStream().

                                                                                                     {
    return run_dir_ + "/" + mergedFileNameStem(ls,stream) + ".dat";
  }

std::string evf::EvFDaqDirector::getOpenDatFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 261 of file EvFDaqDirector.cc.

References outputFileNameStem(), and run_dir_.

                                                                                                   {
    return run_dir_ + "/open/" + outputFileNameStem(ls,stream) + ".dat";
  }

std::string evf::EvFDaqDirector::getOpenRawFilePath	(	const unsigned int	ls,
		const unsigned int	index
	)		const

Definition at line 257 of file EvFDaqDirector.cc.

References bu_run_dir_, and inputFileNameStem().

                                                                                                    {
    return bu_run_dir_ + "/open/" + inputFileNameStem(ls, index) + ".raw";
  }

std::string evf::EvFDaqDirector::getOutputJsonFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 265 of file EvFDaqDirector.cc.

References outputFileNameStem(), and run_dir_.

                                                                                                      {
    return run_dir_ + "/" + outputFileNameStem(ls,stream) + ".jsn";
  }

std::string evf::EvFDaqDirector::getPathForFU

(

)

const

Definition at line 289 of file EvFDaqDirector.cc.

References sm_base_dir_.

                                               {
    return sm_base_dir_;
  }

std::string evf::EvFDaqDirector::getRawFilePath	(	const unsigned int	ls,
		const unsigned int	index
	)		const

Definition at line 253 of file EvFDaqDirector.cc.

References bu_run_dir_, and inputFileNameStem().

Referenced by bumpFile(), and removeFile().

                                                                                                {
    return bu_run_dir_ + "/" + inputFileNameStem(ls, index) + ".raw";
  }

unsigned int evf::EvFDaqDirector::getRunNumber ( ) const

inline

Definition at line 65 of file EvFDaqDirector.h.

References run_.

{ return run_; }

bool evf::EvFDaqDirector::getTestModeNoBuilderUnit ( )

inline

Definition at line 69 of file EvFDaqDirector.h.

References testModeNoBuilderUnit_.

{ return testModeNoBuilderUnit_;}

std::string evf::EvFDaqDirector::initFileName ( std::string const &  stream ) const

private

Definition at line 652 of file EvFDaqDirector.cc.

References run_string_.

Referenced by getInitFilePath().

                                                                      {
    std::stringstream ss;
    ss << run_string_
       << "_" << stream
       << "_pid" << std::setfill('0') << std::setw(5) << getpid()
       << ".ini";
    return ss.str();
  }

std::string evf::EvFDaqDirector::inputFileNameStem ( const unsigned int  ls, const unsigned int  index  ) const

private

Definition at line 626 of file EvFDaqDirector.cc.

References getHLTprescales::index, and run_string_.

Referenced by getJumpFilePath(), getOpenRawFilePath(), and getRawFilePath().

                                                                                                   {
    std::stringstream ss;
    ss << run_string_
       << "_ls" << std::setfill('0') << std::setw(4) << ls
       << "_index" << std::setfill('0') << std::setw(6) << index;
    return ss.str();
  }

FILE * evf::EvFDaqDirector::maybeCreateAndLockFileHeadForStream	(	unsigned int	ls,
		std::string &	stream
	)

Definition at line 606 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_flk, data_rw_stream, and getMergedDatFilePath().

                                                                                              {
    data_rw_stream = fopen(getMergedDatFilePath(ls,stream).c_str(), "a"); //open stream for appending
    data_readwrite_fd_ = fileno(data_rw_stream);
    if (data_readwrite_fd_ == -1)
      edm::LogError("EvFDaqDirector") << "problem with creating filedesc for datamerge "
        << strerror(errno);
    else
      edm::LogInfo("EvFDaqDirector") << "creating filedesc for datamerge "
        << data_readwrite_fd_;
    fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_flk);
    
    return data_rw_stream;
  }

std::string evf::EvFDaqDirector::mergedFileNameStem ( const unsigned int  ls, std::string const &  stream  ) const

private

Definition at line 643 of file EvFDaqDirector.cc.

References hostname_, and run_string_.

Referenced by getMergedDatFilePath().

                                                                                                   {
    std::stringstream ss;
    ss << run_string_
       << "_ls" << std::setfill('0') << std::setw(4) << ls
       << "_" << stream
       << "_" << hostname_;
    return ss.str();
  }

bool evf::EvFDaqDirector::mkFuRunDir ( )

private

std::string& evf::EvFDaqDirector::monitorBaseDir ( )

inline

Definition at line 38 of file EvFDaqDirector.h.

References monitor_base_dir_.

{return monitor_base_dir_;}

void evf::EvFDaqDirector::openFULockfileStream ( std::string &  fuLockFilePath, bool  create  )

private

Definition at line 588 of file EvFDaqDirector.cc.

References fu_readwritelock_fd_, and fu_rw_lock_stream.

Referenced by EvFDaqDirector().

                                                                              {
    if (create) {
      fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
                  S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
    } else {
      fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
    }
    if (fu_readwritelock_fd_ == -1)
      edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock "
        << strerror(errno);
    else
      edm::LogInfo("EvFDaqDirector") << "creating filedesc for fureadwritelock "
        << fu_readwritelock_fd_;

    fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
  }

std::string evf::EvFDaqDirector::outputFileNameStem ( const unsigned int  ls, std::string const &  stream  ) const

private

Definition at line 634 of file EvFDaqDirector.cc.

References run_string_.

Referenced by getOpenDatFilePath(), and getOutputJsonFilePath().

                                                                                                   {
    std::stringstream ss;
    ss << run_string_
       << "_ls" << std::setfill('0') << std::setw(4) << ls
       << "_" << stream
       << "_pid" << std::setfill('0') << std::setw(5) << getpid();
    return ss.str();
  }

void evf::EvFDaqDirector::postEndRun	(	edm::Run const &	run,
		edm::EventSetup const &	es
	)

Definition at line 183 of file EvFDaqDirector.cc.

References bu_base_dir_, bu_readlock_fd_, bu_writelock_fd_, directorBu_, lut2db_cfg::filename, removeFile(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by EvFDaqDirector().

                                                                            {
    close(bu_readlock_fd_);
    close(bu_writelock_fd_);
    if (directorBu_) {
      std::string filename = bu_base_dir_ + "/bu.lock";
      removeFile(filename);
    }
  }

void evf::EvFDaqDirector::preBeginRun	(	edm::RunID const &	id,
		edm::Timestamp const &	ts
	)

Definition at line 192 of file EvFDaqDirector.cc.

References dirManager_, edm::hlt::Exception, evf::DirManager::findHighestRunDir(), DTTTrigCorrFirst::run, run_, and run_dir_.

Referenced by EvFDaqDirector().

                                                                             {

    assert(run_ == id.run());

    // check if run dir exists or make it.
    umask(0);
    int retval = mkdir(run_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error creating run dir "
                      << run_dir_ << " mkdir error:" << strerror(errno) << "\n";
    }

    // check if the requested run is the latest one - issue a warning if it isn't
    if (dirManager_.findHighestRunDir() != run_dir_) {
      edm::LogWarning("EvFDaqDirector") << "DaqDirector Warning checking run dir "
                    << run_dir_ << " this is not the highest run "
                    << dirManager_.findHighestRunDir();
    }


    
  }

int evf::EvFDaqDirector::readBuLock

(

)

Definition at line 416 of file EvFDaqDirector.cc.

References bu_base_dir_, bu_r_flk, bu_r_fulk, bu_r_lock_stream, bu_readlock_fd_, bulock(), CastorDataFrameFilter_impl::check(), evf::DirManager::checkDirEmpty(), gather_cfg::cout, dirManager_, and AlCaHLTBitMon_ParallelJobs::p.

                                 {
    int retval = -1;
    // check if lock file has disappeared and if so whether directory is empty (signal end of run)
    if (!bulock() && dirManager_.checkDirEmpty(bu_base_dir_))
      return retval;
    if (fcntl(bu_readlock_fd_, F_SETLKW, &bu_r_flk) != 0)
      retval = 0;
    if (bu_r_lock_stream) {
      unsigned int readval;
      int check = 0;
      unsigned int *p = &readval;
      check = fseek(bu_r_lock_stream, 0, SEEK_SET);
      if (check == 0) {
    fscanf(bu_r_lock_stream, "%u", p);
    retval = readval;
      }
    } else {
      edm::LogError("EvFDaqDirector") << "error reading bu lock file " << strerror(errno);
      retval = -1;
    }
    fcntl(bu_readlock_fd_, F_SETLKW, &bu_r_fulk);
    std::cout << "readbulock returning " << retval << std::endl;
    return retval;
  }

void evf::EvFDaqDirector::removeFile	(	unsigned int	ls,
		unsigned int	index
	)

Definition at line 300 of file EvFDaqDirector.cc.

References getRawFilePath().

Referenced by postEndRun().

                                                                     {
    removeFile(getRawFilePath(ls,index));
  }

void evf::EvFDaqDirector::removeFile

(

std::string

filename

)

Definition at line 293 of file EvFDaqDirector.cc.

References gather_cfg::cout.

                                                    {
    int retval = remove(filename.c_str());
    if (retval != 0)
      std::cout << "Could not remove used file " << filename << " error "
        << strerror(errno) << "\n";
  }

std::string& evf::EvFDaqDirector::smBaseDir ( )

inline

Definition at line 35 of file EvFDaqDirector.h.

References sm_base_dir_.

{return sm_base_dir_;}

void evf::EvFDaqDirector::tryInitializeFuLockFile

(

)

Definition at line 573 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, getHLTprescales::readIndex(), and testModeNoBuilderUnit_.

Referenced by EvFDaqDirector().

                                               {
    if (fu_rw_lock_stream == 0)
      edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream " 
                      << strerror(errno);
    else {
      edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
      unsigned int readLs = 1, readIndex = 0, jumpLs = 3, jumpIndex = 0;
      if (testModeNoBuilderUnit_)
    fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs, readIndex,
        jumpLs, jumpIndex);
      else
    fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
    }
  }

void evf::EvFDaqDirector::unlockAndCloseMergeStream

(

)

Definition at line 620 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_fulk, and data_rw_stream.

                                                 {
    fflush(data_rw_stream);
    fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_fulk);
    fclose(data_rw_stream);
  }

void evf::EvFDaqDirector::updateBuLock

(

unsigned int

ls

)

Definition at line 304 of file EvFDaqDirector.cc.

References bu_w_flk, bu_w_fulk, bu_w_lock_stream, bu_writelock_fd_, and CastorDataFrameFilter_impl::check().

                                                   {
    int check = 0;
    fcntl(bu_writelock_fd_, F_SETLKW, &bu_w_flk);
    if (bu_w_lock_stream != 0) {
      check = fseek(bu_w_lock_stream, 0, SEEK_SET);
      if (check == 0)
    fprintf(bu_w_lock_stream, "%u", ls);
      else
    edm::LogError("EvFDaqDirector")
      << "seek on bu write lock for updating failed with error "
      << strerror(errno);
    } else
      edm::LogError("EvFDaqDirector") << "bu write lock stream is invalid " << strerror(errno);

    fcntl(bu_writelock_fd_, F_SETLKW, &bu_w_fulk);
  }

bool evf::EvFDaqDirector::updateFuLock	(	unsigned int &	ls,
		std::string &	nextFile,
		bool &	eorSeen
	)

Definition at line 321 of file EvFDaqDirector.cc.

References bu_run_dir_, bumpFile(), CastorDataFrameFilter_impl::check(), fu_readwritelock_fd_, fu_rw_flk, fu_rw_fulk, fu_rw_lock_stream, getEoRFilePath(), jumpIndex_, jumpLS_, O_NONBLOCK, getHLTprescales::readIndex(), AlCaHLTBitMon_QueryRunRegistry::string, testModeNoBuilderUnit_, and TrackValidation_HighPurity_cff::valid.

                                   {
    // close and reopen the stream / fd
    close(fu_readwritelock_fd_);
    std::string fulockfile = bu_run_dir_ + "/fu.lock";
    fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_NONBLOCK, S_IRWXU);
    if (fu_readwritelock_fd_ == -1)
      edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock "
        << strerror(errno);
    else
      edm::LogInfo("EvFDaqDirector") << "created filedesc for fureadwritelock "
        << fu_readwritelock_fd_;
    fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
    edm::LogInfo("EvFDaqDirector") << "Reopened the fw FD & STREAM";

    // obtain lock on the fulock file - this call will block if the lock is held by another process
    int retval = fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_flk);
    //if locking fails just return here 
    if(retval!=0) return false;

    eorSeen = false;
    struct stat buf;
    eorSeen = (stat(getEoRFilePath().c_str(), &buf) == 0);
    bool valid = false;

    // if the stream is readable
    if (fu_rw_lock_stream != 0) {
      unsigned int readLs, readIndex, jumpLs, jumpIndex;
      int check = 0;
      // rewind the stream
      check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
      // if rewinded ok
      if (check == 0) {
    // read its' values
    if (testModeNoBuilderUnit_) 
      fscanf(fu_rw_lock_stream, "%u %u %u %u", &readLs, &readIndex,
         &jumpLs, &jumpIndex);
    else 
      fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);

    // try to bump
    bool bumpedOk = bumpFile(readLs, readIndex, nextFile);

    ls = readLs;
    // there is a new file to grab or lumisection ended
    if (bumpedOk) {
      // rewind and clear
      check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
      if (check == 0) {
        ftruncate(fu_readwritelock_fd_, 0);
        fflush(fu_rw_lock_stream); //this should not be needed ???
      } else
          edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
          << strerror(errno);
      // write new data
      check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
      if (check == 0) {
        // write next index in the file, which is the file the next process should take
        if (testModeNoBuilderUnit_) {
          fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs,
              readIndex + 1, readLs + 2, readIndex + 1);
          jumpLS_ = readLs + 2;
          jumpIndex_ = readIndex;
        } else {
          fprintf(fu_rw_lock_stream, "%u %u", readLs,
              readIndex + 1);
        }
        fflush(fu_rw_lock_stream);
        fsync(fu_readwritelock_fd_);

        valid = true;

        if (testModeNoBuilderUnit_)
          edm::LogInfo("EvFDaqDirector")<< "Written to file: " << readLs << ":"
            << readIndex + 1 << " --> " << readLs + 2
            << ":" << readIndex + 1;
        else
          edm::LogInfo("EvFDaqDirector")<< "Written to file: " << readLs << ":"
            << readIndex + 1;

      } else
          edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
          << strerror(errno) << std::endl;
    }
      } else
    edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error "
                    << strerror(errno) << std::endl;
    } else
      edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);
    //release flock at this point
    fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_fulk);

    return valid;
  }

void evf::EvFDaqDirector::writeDiskAndThrottleStat	(	double	fraction,
		int	highWater,
		int	lowWater
	)

Definition at line 488 of file EvFDaqDirector.cc.

References bu_t_monitor_stream, CastorDataFrameFilter_impl::check(), and gather_cfg::cout.

                                      {
    if (bu_t_monitor_stream != 0) {
      int check = fseek(bu_t_monitor_stream, 0, SEEK_SET);
      if (check == 0)
    fprintf(bu_t_monitor_stream, "%f %d %d", fraction, highWater,
        lowWater);
      else
    std::cout
      << "seek on disk write monitor for updating failed with error "
      << strerror(errno) << std::endl;
    } else
      std::cout << "disk write monitor stream is invalid " << strerror(errno)
        << std::endl;

  }

void evf::EvFDaqDirector::writeLsStatisticsBU	(	unsigned int	ls,
		unsigned int	events,
		unsigned long long	totsize,
		long long	lsusec
	)

Definition at line 469 of file EvFDaqDirector.cc.

References bu_w_monitor_stream, CastorDataFrameFilter_impl::check(), and gather_cfg::cout.

                                                                     {
    if (bu_w_monitor_stream != 0) {
      int check = fseek(bu_w_monitor_stream, 0, SEEK_SET);
      if (check == 0) {
    fprintf(bu_w_monitor_stream, "%u %u %llu %f %f %012lld", ls,
        events, totsize,
        double(totsize) / double(events) / 1000000.,
        double(totsize) / double(lsusec), lsusec);
    fflush(bu_w_monitor_stream);
      } else
    std::cout
      << "seek on bu write monitor for updating failed with error "
      << strerror(errno) << std::endl;
    } else
      std::cout << "bu write monitor stream is invalid " << strerror(errno)
        << std::endl;

  }

void evf::EvFDaqDirector::writeLsStatisticsFU ( unsigned int  ls, unsigned int  events, timeval  completion_time  )

inline

Definition at line 62 of file EvFDaqDirector.h.

{}

Member Data Documentation

std::string evf::EvFDaqDirector::base_dir_

private

Definition at line 93 of file EvFDaqDirector.h.

Referenced by baseDir(), and EvFDaqDirector().

std::string evf::EvFDaqDirector::bu_base_dir_

private

Definition at line 94 of file EvFDaqDirector.h.

Referenced by bulock(), EvFDaqDirector(), fulock(), postEndRun(), and readBuLock().

struct flock evf::EvFDaqDirector::bu_r_flk

private

Definition at line 124 of file EvFDaqDirector.h.

Referenced by readBuLock().

struct flock evf::EvFDaqDirector::bu_r_fulk

private

Definition at line 126 of file EvFDaqDirector.h.

Referenced by readBuLock().

FILE* evf::EvFDaqDirector::bu_r_lock_stream

private

Definition at line 112 of file EvFDaqDirector.h.

Referenced by readBuLock().

int evf::EvFDaqDirector::bu_readlock_fd_

private

Definition at line 106 of file EvFDaqDirector.h.

Referenced by bulock(), postEndRun(), and readBuLock().

std::string evf::EvFDaqDirector::bu_run_dir_

private

Definition at line 103 of file EvFDaqDirector.h.

Referenced by buBaseDir(), EvFDaqDirector(), getEoLSFilePathOnBU(), getEoRFilePath(), getJumpFilePath(), getOpenRawFilePath(), getRawFilePath(), and updateFuLock().

std::string evf::EvFDaqDirector::bu_run_open_dir_

private

Definition at line 104 of file EvFDaqDirector.h.

Referenced by buBaseOpenDir(), and EvFDaqDirector().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream

private

Definition at line 115 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and writeDiskAndThrottleStat().

struct flock evf::EvFDaqDirector::bu_w_flk

private

Definition at line 123 of file EvFDaqDirector.h.

Referenced by updateBuLock().

struct flock evf::EvFDaqDirector::bu_w_fulk

private

Definition at line 125 of file EvFDaqDirector.h.

Referenced by updateBuLock().

FILE* evf::EvFDaqDirector::bu_w_lock_stream

private

Definition at line 111 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateBuLock().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream

private

Definition at line 114 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and writeLsStatisticsBU().

int evf::EvFDaqDirector::bu_writelock_fd_

private

Definition at line 107 of file EvFDaqDirector.h.

Referenced by bulock(), EvFDaqDirector(), postEndRun(), and updateBuLock().

int evf::EvFDaqDirector::data_readwrite_fd_

private

Definition at line 109 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream(), and unlockAndCloseMergeStream().

struct flock evf::EvFDaqDirector::data_rw_flk

private

Definition at line 129 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream().

struct flock evf::EvFDaqDirector::data_rw_fulk

private

Definition at line 130 of file EvFDaqDirector.h.

Referenced by unlockAndCloseMergeStream().

FILE* evf::EvFDaqDirector::data_rw_stream

private

Definition at line 116 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream(), and unlockAndCloseMergeStream().

bool evf::EvFDaqDirector::directorBu_

private

Definition at line 97 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_

private

Definition at line 118 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), findHighestRun(), findHighestRunDir(), preBeginRun(), and readBuLock().

int evf::EvFDaqDirector::fu_readwritelock_fd_

private

Definition at line 108 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), fulock(), openFULockfileStream(), and updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_flk

private

Definition at line 127 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk

private

Definition at line 128 of file EvFDaqDirector.h.

Referenced by updateFuLock().

FILE* evf::EvFDaqDirector::fu_rw_lock_stream

private

Definition at line 113 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), openFULockfileStream(), tryInitializeFuLockFile(), and updateFuLock().

std::string evf::EvFDaqDirector::hostname_

private

Definition at line 100 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and mergedFileNameStem().

unsigned int evf::EvFDaqDirector::jumpIndex_

private

Definition at line 121 of file EvFDaqDirector.h.

Referenced by getJumpFilePath(), getJumpIndex(), and updateFuLock().

unsigned int evf::EvFDaqDirector::jumpLS_

private

Definition at line 121 of file EvFDaqDirector.h.

Referenced by getJumpFilePath(), getJumpLS(), and updateFuLock().

std::string evf::EvFDaqDirector::monitor_base_dir_

private

Definition at line 96 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and monitorBaseDir().

unsigned long evf::EvFDaqDirector::previousFileSize_

private

Definition at line 120 of file EvFDaqDirector.h.

Referenced by bumpFile().

unsigned int evf::EvFDaqDirector::run_

private

Definition at line 98 of file EvFDaqDirector.h.

Referenced by eorFileName(), EvFDaqDirector(), findCurrentRunDir(), getRunNumber(), and preBeginRun().

std::string evf::EvFDaqDirector::run_dir_

private

Definition at line 102 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), fuBaseDir(), getEoLSFilePathOnFU(), getInitFilePath(), getMergedDatFilePath(), getOpenDatFilePath(), getOutputJsonFilePath(), and preBeginRun().

std::string evf::EvFDaqDirector::run_string_

private

Definition at line 101 of file EvFDaqDirector.h.

Referenced by createOutputDirectory(), EvFDaqDirector(), initFileName(), inputFileNameStem(), mergedFileNameStem(), and outputFileNameStem().

std::string evf::EvFDaqDirector::sm_base_dir_

private

Definition at line 95 of file EvFDaqDirector.h.

Referenced by createOutputDirectory(), getPathForFU(), and smBaseDir().

bool evf::EvFDaqDirector::testModeNoBuilderUnit_

private

Definition at line 92 of file EvFDaqDirector.h.

Referenced by bumpFile(), getTestModeNoBuilderUnit(), tryInitializeFuLockFile(), and updateFuLock().