21 #include <boost/lexical_cast.hpp>
22 #include <boost/filesystem/fstream.hpp>
29 struct flock make_flock(short
type, short whence, off_t
start, off_t len, pid_t
pid)
41 testModeNoBuilderUnit_(
42 pset.getUntrackedParameter<bool> (
"testModeNoBuilderUnit",
46 pset.getUntrackedParameter<std::
string> (
"baseDir",
"/data")
49 pset.getUntrackedParameter<std::
string> (
"buBaseDir",
"/data")
52 pset.getUntrackedParameter<bool> (
"directorIsBu",
false)
54 run_(pset.getUntrackedParameter<unsigned int> (
"runNumber",0)),
55 outputAdler32Recheck_(pset.getUntrackedParameter<bool>(
"outputAdler32Recheck",
false)),
56 requireTSPSet_(pset.getUntrackedParameter<bool>(
"requireTransfersPSet",
false)),
57 selectedTransferMode_(pset.getUntrackedParameter<std::
string>(
"selectedTransferMode",
"")),
58 hltSourceDirectory_(pset.getUntrackedParameter<std::
string>(
"hltSourceDirectory",
"")),
59 fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>(
"fuLockPollInterval",2000)),
63 fu_readwritelock_fd_(-1),
64 data_readwrite_fd_(-1),
65 fulocal_rwlock_fd_(-1),
66 fulocal_rwlock_fd2_(-1),
75 dirManager_(base_dir_),
81 bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
82 bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
83 bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
84 bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
85 fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
86 fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
87 data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
88 data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
101 std::stringstream
ss;
102 ss <<
"run" << std::setfill(
'0') << std::setw(6) <<
run_;
108 gethostname(hostname,32);
111 char * fuLockPollIntervalPtr = getenv(
"FFF_LOCKPOLLINTERVAL");
112 if (fuLockPollIntervalPtr) {
121 int retval =
mkdir(
base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
122 if (retval != 0 && errno != EEXIST) {
123 throw cms::Exception(
"DaqDirector") <<
" Error checking for base dir -: "
124 <<
base_dir_ <<
" mkdir error:" << strerror(errno);
130 S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
131 if (retval != 0 && errno != EEXIST) {
132 throw cms::Exception(
"DaqDirector") <<
" Error creating run dir -: "
133 <<
run_dir_ <<
" mkdir error:" << strerror(errno);
141 fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
143 throw cms::Exception(
"DaqDirector") <<
" Error creating/opening a local lock file -: " << fulocal_lock_.c_str() <<
" : " << strerror(errno);
144 chmod(fulocal_lock_.c_str(),0777);
147 fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
149 throw cms::Exception(
"DaqDirector") <<
" Error opening a local lock file -: " << fulocal_lock_.c_str() <<
" : " << strerror(errno);
162 S_IRWXU | S_IRWXG | S_IRWXO);
163 if (retval != 0 && errno != EEXIST) {
166 <<
" mkdir error:" << strerror(errno) <<
"\n";
170 S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
171 if (retval != 0 && errno != EEXIST) {
172 throw cms::Exception(
"DaqDirector") <<
" Error creating bu run open dir -: "
179 O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
181 edm::LogWarning(
"EvFDaqDirector") <<
"problem with creating filedesc for buwritelock -: "
184 edm::LogInfo(
"EvFDaqDirector") <<
"creating filedesc for buwritelock -: "
188 edm::LogWarning(
"EvFDaqDirector")<<
"Error creating write lock stream -: " << strerror(errno);
203 retval =
mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
213 boost::filesystem::rename(tmphltdir,hltdir);
225 if (retval != 0 && errno != EEXIST) {
226 throw cms::Exception(
"DaqDirector") <<
" Error checking for bu base dir -: "
227 <<
bu_base_dir_ <<
" mkdir error:" << strerror(errno) <<
"\n";
281 <<
run_dir_ <<
". This is not the highest run "
291 edm::LogWarning(
"EvFDaqDirector") <<
" Handles to check for files to delete were not set by the input source...";
298 if (it->second->lumi_ == ls) {
300 LogDebug(
"EvFDaqDirector") <<
"Deleting input file -:" << it->second->fileName_;
305 catch (
const boost::filesystem::filesystem_error& ex)
307 edm::LogError(
"EvFDaqDirector") <<
" - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() <<
". Trying again.";
316 edm::LogError(
"EvFDaqDirector") <<
" - deleteFile std::exception CAUGHT -: " << ex.what() <<
". Trying again.";
426 int retval =
remove(filename.c_str());
428 edm::LogError(
"EvFDaqDirector") <<
"Could not remove used file -: " << filename <<
". error = "
440 int lock_attempts = 0;
446 edm::LogWarning(
"EvFDaqDirector") <<
"Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
450 timeval ts_lockbegin;
451 gettimeofday(&ts_lockbegin,0);
459 if (lock_attempts>5000000 || errno==116) {
461 edm::LogWarning(
"EvFDaqDirector") <<
"Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
463 edm::LogWarning(
"EvFDaqDirector") <<
"Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "
464 << errno <<
":"<< strerror(errno) << std::endl;
468 edm::LogWarning(
"EvFDaqDirector") <<
"Detected local EoLS for lumisection "<<
ls ;
480 gettimeofday(&ts_lockend,0);
481 long deltat = (ts_lockend.tv_usec-ts_lockbegin.tv_usec) + (ts_lockend.tv_sec-ts_lockbegin.tv_sec)*1000000;
482 if (deltat>0.) lockWaitTime=deltat;
486 if(retval!=0)
return fileStatus;
490 gettimeofday(&ts_lockend,0);
495 unsigned int readLs,
readIndex, jumpLs, jumpIndex;
504 &jumpLs, &jumpIndex);
511 bool bumpedOk =
bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
522 readIndex + 1, readLs + 2, readIndex + 1);
534 edm::LogInfo(
"EvFDaqDirector") <<
"Written to file -: " << readLs <<
":"
535 << readIndex + 1 <<
" --> " << readLs + 2
536 <<
":" << readIndex + 1;
538 LogDebug(
"EvFDaqDirector") <<
"Written to file -: " << readLs <<
":"
542 throw cms::Exception(
"EvFDaqDirector") <<
"seek on fu read/write lock for updating failed with error "
546 edm::LogError(
"EvFDaqDirector") <<
"seek on fu read/write lock for reading failed with error "
549 edm::LogError(
"EvFDaqDirector") <<
"fu read/write lock stream is invalid " << strerror(errno);
552 timeval ts_preunlock;
553 gettimeofday(&ts_preunlock,0);
554 int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
555 double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
564 if (retvalu==-1)
edm::LogError(
"EvFDaqDirector") <<
"Error unlocking the fu.lock " << strerror(errno);
567 edm::LogDebug(
"EvFDaqDirector") <<
"Waited during lock -: " << locked_period <<
" seconds";
570 if ( fileStatus ==
noFile ) {
575 if (stopFileLS>=0 && (
int)ls > stopFileLS) {
576 edm::LogInfo(
"EvFDaqDirector") <<
"Reached maximum lumisection set by hltd";
585 boost::filesystem::ifstream ij(BUEoLSFile);
589 if (!reader.
parse(ij, deserializeRoot)) {
590 edm::LogError(
"EvFDaqDirector") <<
"Cannot deserialize input JSON file -:" << BUEoLSFile;
605 if (def.find(
'/')==0)
610 if (stat(fullpath.c_str(), &buf) == 0) {
613 DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
618 DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
620 for (
unsigned int i=0;
i<eolsDpd.
getNames().size();
i++)
627 if (def.size()<=1 || def.find(
'/')==std::string::npos) {
631 def = def.substr(def.find(
'/')+1);
638 edm::LogError(
"EvFDaqDirector") <<
" error reading number of files from BU JSON -: " << BUEoLSFile;
641 return boost::lexical_cast<
int>(
data);
651 edm::LogError(
"EvFDaqDirector") <<
" FastMonitoringService not found";
659 if (maxLS>=0 && ls > (
unsigned int)maxLS)
return false;
662 std::stringstream
ss;
663 unsigned int nextIndex =
index;
668 if (stat(nextFile.c_str(), &buf) == 0) {
677 bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
678 unsigned int startingLumi =
ls;
682 if (stat(nextFile.c_str(), &buf) == 0) {
689 if (indexFilesInLS < 0)
694 if ((
int)index<indexFilesInLS) {
696 edm::LogError(
"EvFDaqDirector") <<
"Potential miss of index file in LS -: " << ls <<
". Missing "
697 << nextFile <<
" because " << indexFilesInLS-1 <<
" is the highest index expected. Will not update fu.lock file";
706 if (maxLS>=0 && ls > (
unsigned int)maxLS)
return false;
709 if (stat(nextFile.c_str(), &buf) == 0) {
720 std::string cpCmd =
"cp " + sourceEol +
" " + destEol;
721 edm::LogInfo(
"EvFDaqDirector") <<
" testmode: Running copy cmd -: " << cpCmd;
722 int rc = system(cpCmd.c_str());
724 edm::LogError(
"EvFDaqDirector") <<
" testmode: COPY EOL FAILED!!!!! -: " << cpCmd;
731 eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
740 edm::LogError(
"EvFDaqDirector") <<
"Error creating fu read/write lock stream "
743 edm::LogInfo(
"EvFDaqDirector") <<
"Initializing FU LOCK FILE";
744 unsigned int readLs = 1,
readIndex = 0, jumpLs = 3, jumpIndex = 0;
756 S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
757 chmod(fulockfile.c_str(),0766);
762 edm::LogError(
"EvFDaqDirector") <<
"problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
763 <<
" create:" << create <<
" error:" << strerror(errno);
765 LogDebug(
"EvFDaqDirector") <<
"creating filedesc for fureadwritelock -: "
776 edm::LogError(
"EvFDaqDirector") <<
"problem with creating filedesc for datamerge "
779 LogDebug(
"EvFDaqDirector") <<
"creating filedesc for datamerge -: "
826 if (!boost::filesystem::is_directory(openPath)) {
827 LogDebug(
"EvFDaqDirector") <<
"<open> FU dir not found. Creating... -:" << openPath.string();
828 boost::filesystem::create_directories(openPath);
835 boost::filesystem::ifstream ij(file);
839 if (!reader.
parse(ij, deserializeRoot)) {
840 edm::LogError(
"EvFDaqDirector") <<
"Cannot deserialize input JSON file -:" <<
file;
844 int ret = deserializeRoot.
get(
"lastLS",
"").
asInt();
860 (*transferSystemJson_)[
"destinations"]=destinationsVal;
863 std::vector<std::string> modes = tsPset.
getParameter< std::vector<std::string> >(
"transferModes");
865 (*transferSystemJson_)[
"transferModes"]=modesVal;
867 for (
auto psKeyItr =tsPset.
psetTable().begin();psKeyItr!=tsPset.
psetTable().end(); ++ psKeyItr) {
868 if (psKeyItr->first!=
"destinations" && psKeyItr->first!=
"transferModes") {
871 for (
auto &
mode : modes) {
873 if (!streamDef.
existsAs<std::vector<std::string>>(
mode,
true))
874 throw cms::Exception(
"EvFDaqDirector") <<
" Missing transfer system specification for -:" << psKeyItr->first <<
" (transferMode " <<
mode <<
")";
875 std::vector<std::string> streamDestinations = streamDef.
getParameter<std::vector<std::string>>(
mode);
879 if (!streamDestinations.size())
880 throw cms::Exception(
"EvFDaqDirector") <<
" Missing transter system destination(s) for -: "<< psKeyItr->first <<
", mode:" <<
mode;
882 for (
auto & sdest:streamDestinations) {
883 bool sDestValid=
false;
884 sDestsValue.append(sdest);
885 for (
auto &
dest: destinations) {
886 if (
dest==sdest) sDestValid=
true;
889 throw cms::Exception(
"EvFDaqDirector") <<
" Invalid transter system destination specified for -: "<< psKeyItr->first <<
", mode:" <<
mode <<
", dest:"<<sdest;
891 streamVal[
mode]=sDestsValue;
893 (*transferSystemJson_)[psKeyItr->first] = streamVal;
899 throw cms::Exception(
"EvFDaqDirector") <<
"transferSystem PSet not found";
907 streamRequestName =
stream;
909 std::stringstream
msg;
910 msg <<
"Transfer system mode definitions missing for -: " <<
stream;
914 edm::LogWarning(
"EvFDaqDirector") << msg.str() <<
" (permissive mode)";
920 edm::LogWarning(
"EvFDaqDirector") <<
"Selected mode string is not provided as DaqDirector parameter."
921 <<
"Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
925 throw cms::Exception(
"EvFDaqDirector") <<
"Selected mode string is not provided as DaqDirector parameter.";
930 std::stringstream
msg;
931 msg <<
"Selected transfer mode " <<
selectedTransferMode_ <<
" is not specified for stream " << streamRequestName;
935 edm::LogWarning(
"EvFDaqDirector") << msg.str() <<
" (permissive mode)";
944 if (ret!=
"") ret +=
",";
945 ret+=(*it).asString();
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.
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_
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)
tuple path
else: Piece not in the list, fine.
EvFDaqDirector(const edm::ParameterSet &pset, edm::ActivityRegistry ®)
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
void checkTransferSystemPSet()
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_
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)
psettable const & psetTable() const
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 & getProcessParameterSet()
ParameterSet const & getParameterSet(std::string const &) const
void preGlobalEndLumi(edm::GlobalContext const &globalContext)
bool testModeNoBuilderUnit_
void watchPostGlobalEndRun(PostGlobalEndRun::slot_type const &iSlot)
LuminosityBlockNumber_t luminosityBlock() const
std::string & getDefinition()
evf::FastMonitoringService * fms_
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
char data[epos_bytes_allocation]
void postEndRun(edm::GlobalContext const &globalContext)
Unserialize a JSON document into a Value.
const_iterator end() const
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)