21 #include <boost/lexical_cast.hpp>
22 #include <boost/filesystem/fstream.hpp>
26 using namespace jsoncollector;
31 struct flock make_flock(short
type, short whence, off_t
start, off_t len, pid_t
pid)
44 pset.getUntrackedParameter<std::
string> (
"baseDir",
"/data")
47 pset.getUntrackedParameter<std::
string> (
"buBaseDir",
"/data")
50 pset.getUntrackedParameter<bool> (
"directorIsBu",
false)
52 run_(pset.getUntrackedParameter<unsigned int> (
"runNumber",0)),
53 outputAdler32Recheck_(pset.getUntrackedParameter<bool>(
"outputAdler32Recheck",
false)),
54 requireTSPSet_(pset.getUntrackedParameter<bool>(
"requireTransfersPSet",
false)),
55 selectedTransferMode_(pset.getUntrackedParameter<std::
string>(
"selectedTransferMode",
"")),
56 hltSourceDirectory_(pset.getUntrackedParameter<std::
string>(
"hltSourceDirectory",
"")),
57 fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>(
"fuLockPollInterval",2000)),
58 emptyLumisectionMode_(pset.getUntrackedParameter<bool>(
"emptyLumisectionMode",
false)),
62 fu_readwritelock_fd_(-1),
63 data_readwrite_fd_(-1),
64 fulocal_rwlock_fd_(-1),
65 fulocal_rwlock_fd2_(-1),
74 dirManager_(base_dir_),
78 bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
79 bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
80 bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
81 bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
82 fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
83 fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
84 data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
85 data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
100 ss <<
"run" << std::setfill(
'0') << std::setw(6) <<
run_;
106 gethostname(hostname,32);
109 char * fuLockPollIntervalPtr = getenv(
"FFF_LOCKPOLLINTERVAL");
110 if (fuLockPollIntervalPtr) {
118 char * emptyLumiModePtr = getenv(
"FFF_EMPTYLSMODE");
119 if (emptyLumiModePtr) {
125 int retval = mkdir(
base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
126 if (retval != 0 && errno != EEXIST) {
127 throw cms::Exception(
"DaqDirector") <<
" Error checking for base dir -: "
128 <<
base_dir_ <<
" mkdir error:" << strerror(errno);
134 S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
135 if (retval != 0 && errno != EEXIST) {
136 throw cms::Exception(
"DaqDirector") <<
" Error creating run dir -: "
137 <<
run_dir_ <<
" mkdir error:" << strerror(errno);
145 fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
147 throw cms::Exception(
"DaqDirector") <<
" Error creating/opening a local lock file -: " << fulocal_lock_.c_str() <<
" : " << strerror(errno);
148 chmod(fulocal_lock_.c_str(),0777);
151 fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
153 throw cms::Exception(
"DaqDirector") <<
" Error opening a local lock file -: " << fulocal_lock_.c_str() <<
" : " << strerror(errno);
166 S_IRWXU | S_IRWXG | S_IRWXO);
167 if (retval != 0 && errno != EEXIST) {
170 <<
" mkdir error:" << strerror(errno) <<
"\n";
174 S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
175 if (retval != 0 && errno != EEXIST) {
176 throw cms::Exception(
"DaqDirector") <<
" Error creating bu run open dir -: "
183 O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
185 edm::LogWarning(
"EvFDaqDirector") <<
"problem with creating filedesc for buwritelock -: "
188 edm::LogInfo(
"EvFDaqDirector") <<
"creating filedesc for buwritelock -: "
192 edm::LogWarning(
"EvFDaqDirector")<<
"Error creating write lock stream -: " << strerror(errno);
207 retval = mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
209 boost::filesystem::copy_file(
hltSourceDirectory_+
"/HltConfig.py",tmphltdir+
"/HltConfig.py");
211 boost::filesystem::copy_file(
hltSourceDirectory_+
"/CMSSW_VERSION",tmphltdir+
"/CMSSW_VERSION");
215 boost::filesystem::copy_file(
hltSourceDirectory_+
"/fffParameters.jsn",tmphltdir+
"/fffParameters.jsn");
217 boost::filesystem::rename(tmphltdir,hltdir);
228 retval = mkdir(
bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
229 if (retval != 0 && errno != EEXIST) {
230 throw cms::Exception(
"DaqDirector") <<
" Error checking for bu base dir -: "
231 <<
bu_base_dir_ <<
" mkdir error:" << strerror(errno) <<
"\n";
288 <<
run_dir_ <<
". This is not the highest run "
298 edm::LogWarning(
"EvFDaqDirector") <<
" Handles to check for files to delete were not set by the input source...";
305 if (it->second->lumi_ == ls) {
307 LogDebug(
"EvFDaqDirector") <<
"Deleting input file -:" << it->second->fileName_;
312 catch (
const boost::filesystem::filesystem_error& ex)
314 edm::LogError(
"EvFDaqDirector") <<
" - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() <<
". Trying again.";
323 edm::LogError(
"EvFDaqDirector") <<
" - deleteFile std::exception CAUGHT -: " << ex.what() <<
". Trying again.";
433 int retval =
remove(filename.c_str());
435 edm::LogError(
"EvFDaqDirector") <<
"Could not remove used file -: " << filename <<
". error = "
447 int lock_attempts = 0;
453 edm::LogWarning(
"EvFDaqDirector") <<
"Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
457 timeval ts_lockbegin;
458 gettimeofday(&ts_lockbegin,0);
466 if (lock_attempts>5000000 || errno==116) {
468 edm::LogWarning(
"EvFDaqDirector") <<
"Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
470 edm::LogWarning(
"EvFDaqDirector") <<
"Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "
471 << errno <<
":"<< strerror(errno) << std::endl;
475 edm::LogWarning(
"EvFDaqDirector") <<
"Detected local EoLS for lumisection "<<
ls ;
487 gettimeofday(&ts_lockend,0);
488 long deltat = (ts_lockend.tv_usec-ts_lockbegin.tv_usec) + (ts_lockend.tv_sec-ts_lockbegin.tv_sec)*1000000;
489 if (deltat>0.) lockWaitTime=deltat;
493 if(retval!=0)
return fileStatus;
497 gettimeofday(&ts_lockend,0);
512 unsigned int currentLs = readLs;
513 bool bumpedOk =
false;
516 if (ls && ls+1 < currentLs) ls++;
519 bumpedOk =
bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
521 if (ls && readLs>currentLs && currentLs > ls) {
529 if (ls==0 && readLs>currentLs) {
551 LogDebug(
"EvFDaqDirector") <<
"Written to file -: " << readLs <<
":" << readIndex + 1;
554 throw cms::Exception(
"EvFDaqDirector") <<
"seek on fu read/write lock for updating failed with error " << strerror(errno);
557 else if (currentLs < readLs) {
566 LogDebug(
"EvFDaqDirector") <<
"Written to file -: " << readLs <<
":" <<
readIndex;
569 throw cms::Exception(
"EvFDaqDirector") <<
"seek on fu read/write lock for updating failed with error " << strerror(errno);
573 edm::LogError(
"EvFDaqDirector") <<
"seek on fu read/write lock for reading failed with error " << strerror(errno);
576 edm::LogError(
"EvFDaqDirector") <<
"fu read/write lock stream is invalid " << strerror(errno);
580 timeval ts_preunlock;
581 gettimeofday(&ts_preunlock,0);
582 int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
583 double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
592 if (retvalu==-1)
edm::LogError(
"EvFDaqDirector") <<
"Error unlocking the fu.lock " << strerror(errno);
595 edm::LogDebug(
"EvFDaqDirector") <<
"Waited during lock -: " << locked_period <<
" seconds";
598 if ( fileStatus ==
noFile ) {
603 if (stopFileLS>=0 && (
int)ls > stopFileLS) {
604 edm::LogInfo(
"EvFDaqDirector") <<
"Reached maximum lumisection set by hltd";
613 boost::filesystem::ifstream ij(BUEoLSFile);
617 if (!reader.
parse(ij, deserializeRoot)) {
618 edm::LogError(
"EvFDaqDirector") <<
"Cannot deserialize input JSON file -:" << BUEoLSFile;
633 if (def.find(
'/')==0)
638 if (stat(fullpath.c_str(), &buf) == 0) {
641 DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
646 DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
648 for (
unsigned int i=0;
i<eolsDpd.
getNames().size();
i++)
655 if (def.size()<=1 || def.find(
'/')==std::string::npos) {
659 def = def.substr(def.find(
'/')+1);
666 edm::LogError(
"EvFDaqDirector") <<
" error reading number of files from BU JSON -: " << BUEoLSFile;
669 return boost::lexical_cast<
int>(
data);
679 edm::LogError(
"EvFDaqDirector") <<
" FastMonitoringService not found";
687 if (maxLS>=0 && ls > (
unsigned int)maxLS)
return false;
690 std::stringstream
ss;
691 unsigned int nextIndex =
index;
696 if (stat(nextFile.c_str(), &buf) == 0) {
705 bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
709 if (stat(nextFile.c_str(), &buf) == 0) {
716 if (indexFilesInLS < 0)
721 if ((
int)index<indexFilesInLS) {
723 edm::LogError(
"EvFDaqDirector") <<
"Potential miss of index file in LS -: " << ls <<
". Missing "
724 << nextFile <<
" because " << indexFilesInLS-1 <<
" is the highest index expected. Will not update fu.lock file";
733 if (maxLS>=0 && ls > (
unsigned int)maxLS)
return false;
736 if (stat(nextFile.c_str(), &buf) == 0) {
747 eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
756 edm::LogError(
"EvFDaqDirector") <<
"Error creating fu read/write lock stream "
759 edm::LogInfo(
"EvFDaqDirector") <<
"Initializing FU LOCK FILE";
768 S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
769 chmod(fulockfile.c_str(),0766);
774 edm::LogError(
"EvFDaqDirector") <<
"problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
775 <<
" create:" << create <<
" error:" << strerror(errno);
777 LogDebug(
"EvFDaqDirector") <<
"creating filedesc for fureadwritelock -: "
788 edm::LogError(
"EvFDaqDirector") <<
"problem with creating filedesc for datamerge "
791 LogDebug(
"EvFDaqDirector") <<
"creating filedesc for datamerge -: "
838 if (!boost::filesystem::is_directory(openPath)) {
839 LogDebug(
"EvFDaqDirector") <<
"<open> FU dir not found. Creating... -:" << openPath.string();
840 boost::filesystem::create_directories(openPath);
847 boost::filesystem::ifstream ij(file);
851 if (!reader.
parse(ij, deserializeRoot)) {
852 edm::LogError(
"EvFDaqDirector") <<
"Cannot deserialize input JSON file -:" <<
file;
856 int ret = deserializeRoot.
get(
"lastLS",
"").
asInt();
873 std::vector<std::string>
destinations = tsPset.getParameter<std::vector<std::string>>(
"destinations");
874 for (
auto &
dest: destinations) destinationsVal.append(
dest);
875 (*transferSystemJson_)[
"destinations"]=destinationsVal;
878 std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >(
"transferModes");
880 (*transferSystemJson_)[
"transferModes"]=modesVal;
882 for (
auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
883 if (psKeyItr->first!=
"destinations" && psKeyItr->first!=
"transferModes") {
886 for (
auto &
mode : modes) {
888 if (!streamDef.
existsAs<std::vector<std::string>>(
mode,
true))
889 throw cms::Exception(
"EvFDaqDirector") <<
" Missing transfer system specification for -:" << psKeyItr->first <<
" (transferMode " <<
mode <<
")";
890 std::vector<std::string> streamDestinations = streamDef.
getParameter<std::vector<std::string>>(
mode);
894 if (!streamDestinations.size())
895 throw cms::Exception(
"EvFDaqDirector") <<
" Missing transter system destination(s) for -: "<< psKeyItr->first <<
", mode:" <<
mode;
897 for (
auto & sdest:streamDestinations) {
898 bool sDestValid=
false;
899 sDestsValue.append(sdest);
900 for (
auto &
dest: destinations) {
901 if (
dest==sdest) sDestValid=
true;
904 throw cms::Exception(
"EvFDaqDirector") <<
" Invalid transter system destination specified for -: "<< psKeyItr->first <<
", mode:" <<
mode <<
", dest:"<<sdest;
906 streamVal[
mode]=sDestsValue;
908 (*transferSystemJson_)[psKeyItr->first] = streamVal;
914 throw cms::Exception(
"EvFDaqDirector") <<
"transferSystem PSet not found";
922 streamRequestName = stream;
924 std::stringstream
msg;
925 msg <<
"Transfer system mode definitions missing for -: " << stream;
929 edm::LogWarning(
"EvFDaqDirector") << msg.str() <<
" (permissive mode)";
935 edm::LogWarning(
"EvFDaqDirector") <<
"Selected mode string is not provided as DaqDirector parameter."
936 <<
"Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
940 throw cms::Exception(
"EvFDaqDirector") <<
"Selected mode string is not provided as DaqDirector parameter.";
945 std::stringstream
msg;
946 msg <<
"Selected transfer mode " <<
selectedTransferMode_ <<
" is not specified for stream " << streamRequestName;
950 edm::LogWarning(
"EvFDaqDirector") << msg.str() <<
" (permissive mode)";
959 if (ret!=
"") ret +=
",";
960 ret+=(*it).asString();
967 int proc_flag_fd = open(proc_flag.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
unsigned int maxNumberOfThreads() const
T getParameter(std::string const &) const
std::string getStreamDestinations(std::string const &stream) const
Value get(UInt index, const Value &defaultValue) const
std::vector< std::string > & getData()
std::string protocolBufferHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
tuple start
Check for commandline option errors.
std::string bolsFileName(const unsigned int run, const unsigned int ls)
std::string streamerDataChecksumFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string getMergedProtocolBufferHistogramFilePath(const unsigned int ls, std::string const &stream) const
void watchPreallocate(Preallocate::slot_type const &iSlot)
std::vector< int > streamFileTracker_
const_iterator begin() const
bool existsAs(std::string const ¶meterName, bool trackiness=true) const
checks if a parameter exists as a given type
std::list< std::pair< int, InputFile * > > * filesToDeletePtr_
std::string getMergedDatChecksumFilePath(const unsigned int ls, std::string const &stream) const
std::shared_ptr< Json::Value > transferSystemJson_
void openFULockfileStream(std::string &fuLockFilePath, bool create)
void accumulateFileSize(unsigned int lumi, unsigned long fileSize)
void watchPreGlobalEndLumi(PreGlobalEndLumi::slot_type const &iSlot)
std::string getInitFilePath(std::string const &stream) const
std::string getMergedRootHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::string inputRawFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
ParameterSet const & getParameterSet(ParameterSetID const &id)
pthread_mutex_t init_lock_
LuminosityBlockID const & luminosityBlockID() const
std::string getProtocolBufferHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::string getRawFilePath(const unsigned int ls, const unsigned int index) const
std::string getOpenInitFilePath(std::string const &stream) const
std::string getOpenRawFilePath(const unsigned int ls, const unsigned int index) const
Value & append(const Value &value)
Append value to array at the end.
void createProcessingNotificationMaybe() const
bool parse(const std::string &document, Value &root, bool collectComments=true)
Read a Value from a JSON document.
std::string getEoLSFilePathOnBU(const unsigned int ls) const
std::string getEoRFilePath() const
unsigned long previousFileSize_
ParameterSetID const & parameterSetID() const
void createRunOpendirMaybe()
unsigned int maxNumberOfStreams() const
std::string hltSourceDirectory_
std::string getOpenProtocolBufferHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::mutex * fileDeleteLockPtr_
std::string streamerDataFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string getMergedDatFilePath(const unsigned int ls, std::string const &stream) const
bool check(const std::string &)
FileStatus updateFuLock(unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
std::string getOpenOutputJsonFilePath(const unsigned int ls, std::string const &stream) const
std::string getOpenRootHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::string getRootHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::string getOpenDatFilePath(const unsigned int ls, std::string const &stream) const
std::string inputJsonFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
std::string stopFilePath_
FILE * maybeCreateAndLockFileHeadForStream(unsigned int ls, std::string &stream)
void unlockAndCloseMergeStream()
void tryInitializeFuLockFile()
void removeFile(unsigned int ls, unsigned int index)
std::string selectedTransferMode_
std::string streamerJsonFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string getOpenInputJsonFilePath(const unsigned int ls, const unsigned int index) const
std::string eorFileName(const unsigned int run)
int readLastLSEntry(std::string const &file)
virtual void deserialize(Json::Value &root)
unsigned int eolsNFilesIndex_
bool emptyLumisectionMode_
void preBeginJob(edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &)
void watchPreGlobalBeginRun(PreGlobalBeginRun::slot_type const &iSlot)
std::string getOutputJsonFilePath(const unsigned int ls, std::string const &stream) const
int getNFilesFromEoLS(std::string BUEoLSFile)
void preBeginRun(edm::GlobalContext const &globalContext)
std::string rootHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
void preSourceEvent(edm::StreamID const &streamID)
std::string initFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
unsigned long long uint64_t
ParameterSet const & getParameterSet(std::string const &) const
void preGlobalEndLumi(edm::GlobalContext const &globalContext)
void watchPostGlobalEndRun(PostGlobalEndRun::slot_type const &iSlot)
LuminosityBlockNumber_t luminosityBlock() const
std::string & getDefinition()
evf::FastMonitoringService * fms_
void watchPreBeginJob(PreBeginJob::slot_type const &iSlot)
convenience function for attaching to signal
std::string rootHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string findHighestRunDir()
std::string getBoLSFilePathOnFU(const unsigned int ls) const
void postEndRun(edm::GlobalContext const &globalContext)
Unserialize a JSON document into a Value.
const_iterator end() const
void checkTransferSystemPSet(edm::ProcessContext const &pc)
std::vector< std::string > const & getNames()
std::string getEoLSFilePathOnFU(const unsigned int ls) const
void watchPreSourceEvent(PreSourceEvent::slot_type const &iSlot)
bool bumpFile(unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize, int maxLS)
volatile std::atomic< bool > shutdown_flag false
void preallocate(edm::service::SystemBounds const &bounds)
struct flock data_rw_fulk
Iterator for object and array value.
std::string getInputJsonFilePath(const unsigned int ls, const unsigned int index) const
std::string getRunOpenDirPath() const
std::string protocolBufferHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
unsigned int fuLockPollInterval_
std::string eolsFileName(const unsigned int run, const unsigned int ls)
std::string streamerDataFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string getEoRFilePathOnFU() const
std::string bu_run_open_dir_
array value (ordered list)