22 #include <boost/lexical_cast.hpp> 23 #include <boost/filesystem/fstream.hpp> 33 const std::vector<std::string> EvFDaqDirector::MergeTypeNames_ = {
"",
"DAT",
"PB",
"JSNDATA"};
37 base_dir_(pset.getUntrackedParameter<
std::
string> (
"baseDir",
".")),
38 bu_base_dir_(pset.getUntrackedParameter<
std::
string> (
"buBaseDir",
".")),
39 directorBu_(pset.getUntrackedParameter<
bool> (
"directorIsBu",
false)),
40 run_(pset.getUntrackedParameter<unsigned
int> (
"runNumber",0)),
41 outputAdler32Recheck_(pset.getUntrackedParameter<
bool>(
"outputAdler32Recheck",
false)),
42 requireTSPSet_(pset.getUntrackedParameter<
bool>(
"requireTransfersPSet",
false)),
43 selectedTransferMode_(pset.getUntrackedParameter<
std::
string>(
"selectedTransferMode",
"")),
44 hltSourceDirectory_(pset.getUntrackedParameter<
std::
string>(
"hltSourceDirectory",
"")),
45 fuLockPollInterval_(pset.getUntrackedParameter<unsigned
int>(
"fuLockPollInterval",2000)),
46 mergeTypePset_(pset.getUntrackedParameter<
std::
string>(
"mergeTypePset",
"")),
50 fu_readwritelock_fd_(-1),
51 fulocal_rwlock_fd_(-1),
52 fulocal_rwlock_fd2_(-1),
56 dirManager_(base_dir_),
58 bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
59 bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
60 bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
61 bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
62 fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
63 fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
74 gethostname(hostname,32);
77 char * fuLockPollIntervalPtr = getenv(
"FFF_LOCKPOLLINTERVAL");
78 if (fuLockPollIntervalPtr) {
83 catch( boost::bad_lexical_cast
const& ) {
93 ss <<
"run" << std::setfill(
'0') << std::setw(6) <<
run_;
98 int retval =
mkdir(
base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
99 if (retval != 0 && errno != EEXIST) {
100 throw cms::Exception(
"DaqDirector") <<
" Error checking for base dir -: " 101 <<
base_dir_ <<
" mkdir error:" << strerror(errno);
107 S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
108 if (retval != 0 && errno != EEXIST) {
109 throw cms::Exception(
"DaqDirector") <<
" Error creating run dir -: " 110 <<
run_dir_ <<
" mkdir error:" << strerror(errno);
118 fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
120 throw cms::Exception(
"DaqDirector") <<
" Error creating/opening a local lock file -: " << fulocal_lock_.c_str() <<
" : " << strerror(errno);
121 chmod(fulocal_lock_.c_str(),0777);
124 fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
126 throw cms::Exception(
"DaqDirector") <<
" Error opening a local lock file -: " << fulocal_lock_.c_str() <<
" : " << strerror(errno);
139 S_IRWXU | S_IRWXG | S_IRWXO);
140 if (retval != 0 && errno != EEXIST) {
143 <<
" mkdir error:" << strerror(errno) <<
"\n";
147 S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
148 if (retval != 0 && errno != EEXIST) {
149 throw cms::Exception(
"DaqDirector") <<
" Error creating bu run open dir -: " 156 O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
158 edm::LogWarning(
"EvFDaqDirector") <<
"problem with creating filedesc for buwritelock -: " 161 edm::LogInfo(
"EvFDaqDirector") <<
"creating filedesc for buwritelock -: " 165 edm::LogWarning(
"EvFDaqDirector")<<
"Error creating write lock stream -: " << strerror(errno);
180 retval =
mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
181 if (retval != 0 && errno != EEXIST)
183 <<
" Error creating bu run dir -: " << hltdir
184 <<
" mkdir error:" << strerror(errno) <<
"\n";
186 boost::filesystem::copy_file(
hltSourceDirectory_+
"/HltConfig.py",tmphltdir+
"/HltConfig.py");
188 boost::filesystem::copy_file(
hltSourceDirectory_+
"/fffParameters.jsn",tmphltdir+
"/fffParameters.jsn");
190 boost::filesystem::rename(tmphltdir,hltdir);
202 if (retval != 0 && errno != EEXIST) {
203 throw cms::Exception(
"DaqDirector") <<
" Error checking for bu base dir -: " 204 <<
bu_base_dir_ <<
" mkdir error:" << strerror(errno) <<
"\n";
215 std::stringstream sstp;
246 desc.
setComment(
"Service used for file locking arbitration and for propagating information between other EvF components");
249 desc.
addUntracked<
unsigned int> (
"runNumber",0)->setComment(
"Run Number in ramdisk to open");
250 desc.
addUntracked<
bool>(
"outputAdler32Recheck",
false)->setComment(
"Check Adler32 of per-process output files while micro-merging");
251 desc.
addUntracked<
bool>(
"requireTransfersPSet",
false)->setComment(
"Require complete transferSystem PSet in the process configuration");
252 desc.
addUntracked<
std::string>(
"selectedTransferMode",
"")->setComment(
"Selected transfer mode (choice in Lvl0 propagated as Python parameter");
253 desc.
addUntracked<
unsigned int>(
"fuLockPollInterval",2000)->setComment(
"Lock polling interval in microseconds for the input directory file lock");
254 desc.
addUntracked<
std::string>(
"mergingPset",
"")->setComment(
"Name of merging PSet to look for merging type definitions for streams");
256 descriptions.
add(
"EvFDaqDirector", desc);
272 <<
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.";
312 catch (
const boost::filesystem::filesystem_error&) {}
316 edm::LogError(
"EvFDaqDirector") <<
" - deleteFile std::exception CAUGHT -: " << ex.what() <<
". Trying again.";
424 int retval =
remove(filename.c_str());
426 edm::LogError(
"EvFDaqDirector") <<
"Could not remove used file -: " << filename <<
". error = " 438 int lock_attempts = 0;
439 long total_lock_attempts = 0;
445 if (stopFileCheck==0 || stopFilePidCheck==0) {
446 if (stopFileCheck==0)
455 edm::LogWarning(
"EvFDaqDirector") <<
"Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
461 timeval ts_lockbegin;
462 gettimeofday(&ts_lockbegin,
nullptr);
471 if (lock_attempts>5000000 || errno==116) {
473 edm::LogWarning(
"EvFDaqDirector") <<
"Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
475 edm::LogWarning(
"EvFDaqDirector") <<
"Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno " 476 << errno <<
":"<< strerror(errno) << std::endl;
480 edm::LogWarning(
"EvFDaqDirector") <<
"Detected local EoLS for lumisection "<<
ls ;
490 if (total_lock_attempts>5*60000000) {
491 edm::LogError(
"EvFDaqDirector") <<
"Unable to obtain a lock for 5 minutes. Stopping polling activity.";
497 gettimeofday(&ts_lockend,
nullptr);
498 long deltat = (ts_lockend.tv_usec-ts_lockbegin.tv_usec) + (ts_lockend.tv_sec-ts_lockbegin.tv_sec)*1000000;
499 if (deltat>0.) lockWaitTime=deltat;
501 if(retval!=0)
return fileStatus;
505 gettimeofday(&ts_lockend,0);
509 int fu_readwritelock_fd2 = open(
fulockfile_.c_str(), O_RDWR, S_IRWXU);
510 if (fu_readwritelock_fd2 == -1)
512 <<
" create. error:" << strerror(errno);
514 FILE * fu_rw_lock_stream2 = fdopen(fu_readwritelock_fd2,
"r+");
517 if (fu_rw_lock_stream2 !=
nullptr) {
521 check = fseek(fu_rw_lock_stream2, 0, SEEK_SET);
525 fscanf(fu_rw_lock_stream2,
"%u %u", &readLs, &readIndex);
528 unsigned int currentLs = readLs;
529 bool bumpedOk =
false;
532 if (ls && ls+1 < currentLs) ls++;
535 bumpedOk =
bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
537 if (ls && readLs>currentLs && currentLs > ls) {
545 if (ls==0 && readLs>currentLs) {
559 check = fseek(fu_rw_lock_stream2, 0, SEEK_SET);
561 ftruncate(fu_readwritelock_fd2, 0);
563 fprintf(fu_rw_lock_stream2,
"%u %u", readLs, readIndex + 1);
564 fflush(fu_rw_lock_stream2);
565 fsync(fu_readwritelock_fd2);
567 LogDebug(
"EvFDaqDirector") <<
"Written to file -: " << readLs <<
":" << readIndex + 1;
570 throw cms::Exception(
"EvFDaqDirector") <<
"seek on fu read/write lock for updating failed with error " << strerror(errno);
573 else if (currentLs < readLs) {
575 check = fseek(fu_rw_lock_stream2, 0, SEEK_SET);
577 ftruncate(fu_readwritelock_fd2, 0);
579 fprintf(fu_rw_lock_stream2,
"%u %u", readLs, readIndex);
580 fflush(fu_rw_lock_stream2);
581 fsync(fu_readwritelock_fd2);
582 LogDebug(
"EvFDaqDirector") <<
"Written to file -: " << readLs <<
":" <<
readIndex;
585 throw cms::Exception(
"EvFDaqDirector") <<
"seek on fu read/write lock for updating failed with error " << strerror(errno);
589 edm::LogError(
"EvFDaqDirector") <<
"seek on fu read/write lock for reading failed with error " << strerror(errno);
592 edm::LogError(
"EvFDaqDirector") <<
"fu read/write lock stream is invalid " << strerror(errno);
594 fclose(fu_rw_lock_stream2);
597 timeval ts_preunlock;
598 gettimeofday(&ts_preunlock,0);
599 int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
600 double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
609 if (retvalu==-1)
edm::LogError(
"EvFDaqDirector") <<
"Error unlocking the fu.lock " << strerror(errno);
612 edm::LogDebug(
"EvFDaqDirector") <<
"Waited during lock -: " << locked_period <<
" seconds";
615 if ( fileStatus ==
noFile ) {
620 if (stopFileLS>=0 && (
int)ls > stopFileLS) {
621 edm::LogInfo(
"EvFDaqDirector") <<
"Reached maximum lumisection set by hltd";
630 boost::filesystem::ifstream ij(BUEoLSFile);
634 if (!reader.
parse(ij, deserializeRoot)) {
635 edm::LogError(
"EvFDaqDirector") <<
"Cannot deserialize input JSON file -:" << BUEoLSFile;
648 while (!def.empty()) {
650 if (def.find(
'/')==0)
655 if (
stat(fullpath.c_str(), &buf) == 0) {
658 DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
663 DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
665 for (
unsigned int i=0;
i<eolsDpd.
getNames().size();
i++)
672 if (def.size()<=1 || def.find(
'/')==std::string::npos) {
676 def = def.substr(def.find(
'/')+1);
683 edm::LogError(
"EvFDaqDirector") <<
" error reading number of files from BU JSON -: " << BUEoLSFile;
686 return boost::lexical_cast<
int>(
data);
700 if (maxLS>=0 && ls > (
unsigned int)maxLS)
return false;
703 std::stringstream ss;
704 unsigned int nextIndex =
index;
709 if (
stat(nextFile.c_str(), &buf) == 0) {
718 bool eolFound = (
stat(BUEoLSFile.c_str(), &buf) == 0);
722 if (
stat(nextFile.c_str(), &buf) == 0) {
729 if (indexFilesInLS < 0)
734 if ((
int)index<indexFilesInLS) {
736 edm::LogError(
"EvFDaqDirector") <<
"Potential miss of index file in LS -: " << ls <<
". Missing " 737 << nextFile <<
" because " << indexFilesInLS-1 <<
" is the highest index expected. Will not update fu.lock file";
746 if (maxLS>=0 && ls > (
unsigned int)maxLS)
return false;
749 if (
stat(nextFile.c_str(), &buf) == 0) {
760 eolFound = (
stat(BUEoLSFile.c_str(), &buf) == 0);
769 edm::LogError(
"EvFDaqDirector") <<
"Error creating fu read/write lock stream " 772 edm::LogInfo(
"EvFDaqDirector") <<
"Initializing FU LOCK FILE";
781 S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
788 <<
" create:" << create <<
" error:" << strerror(errno);
790 LogDebug(
"EvFDaqDirector") <<
"creating filedesc for fureadwritelock -: " 795 edm::LogError(
"EvFDaqDirector") <<
"problem with opening fuwritelock file stream -: " << strerror(errno);
833 if (!boost::filesystem::is_directory(openPath)) {
834 LogDebug(
"EvFDaqDirector") <<
"<open> FU dir not found. Creating... -:" << openPath.string();
835 boost::filesystem::create_directories(openPath);
842 boost::filesystem::ifstream ij(file);
846 if (!reader.
parse(ij, deserializeRoot)) {
847 edm::LogError(
"EvFDaqDirector") <<
"Cannot deserialize input JSON file -:" <<
file;
851 int ret = deserializeRoot.
get(
"lastLS",
"").
asInt();
868 std::vector<std::string>
destinations = tsPset.getParameter<std::vector<std::string>>(
"destinations");
869 for (
auto &
dest: destinations) destinationsVal.append(
dest);
870 (*transferSystemJson_)[
"destinations"]=destinationsVal;
873 std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >(
"transferModes");
875 (*transferSystemJson_)[
"transferModes"]=modesVal;
877 for (
auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
878 if (psKeyItr->first!=
"destinations" && psKeyItr->first!=
"transferModes") {
881 for (
auto &
mode : modes) {
883 if (!streamDef.
existsAs<std::vector<std::string>>(
mode,
true))
884 throw cms::Exception(
"EvFDaqDirector") <<
" Missing transfer system specification for -:" << psKeyItr->first <<
" (transferMode " <<
mode <<
")";
885 std::vector<std::string> streamDestinations = streamDef.
getParameter<std::vector<std::string>>(
mode);
889 if (streamDestinations.empty())
890 throw cms::Exception(
"EvFDaqDirector") <<
" Missing transter system destination(s) for -: "<< psKeyItr->first <<
", mode:" <<
mode;
892 for (
auto & sdest:streamDestinations) {
893 bool sDestValid=
false;
894 sDestsValue.append(sdest);
895 for (
auto &
dest: destinations) {
896 if (
dest==sdest) sDestValid=
true;
899 throw cms::Exception(
"EvFDaqDirector") <<
" Invalid transter system destination specified for -: "<< psKeyItr->first <<
", mode:" <<
mode <<
", dest:"<<sdest;
901 streamVal[
mode]=sDestsValue;
903 (*transferSystemJson_)[psKeyItr->first] = streamVal;
909 throw cms::Exception(
"EvFDaqDirector") <<
"transferSystem PSet not found";
917 streamRequestName = stream;
919 std::stringstream
msg;
920 msg <<
"Transfer system mode definitions missing for -: " << stream;
924 edm::LogWarning(
"EvFDaqDirector") << msg.str() <<
" (permissive mode)";
930 edm::LogWarning(
"EvFDaqDirector") <<
"Selected mode string is not provided as DaqDirector parameter." 931 <<
"Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
935 throw cms::Exception(
"EvFDaqDirector") <<
"Selected mode string is not provided as DaqDirector parameter.";
940 std::stringstream
msg;
941 msg <<
"Selected transfer mode " <<
selectedTransferMode_ <<
" is not specified for stream " << streamRequestName;
945 edm::LogWarning(
"EvFDaqDirector") << msg.str() <<
" (permissive mode)";
954 if (!ret.empty()) ret +=
",";
955 ret+=(*it).asString();
970 tbb::concurrent_hash_map<std::string,std::string>::accessor ac;
972 ac->second = streamType;
980 tbb::concurrent_hash_map<std::string,std::string>::const_accessor search_ac;
982 return search_ac->second;
984 edm::LogInfo(
"EvFDaqDirector") <<
" No merging type specified for stream " << stream <<
". Using default value";
986 tbb::concurrent_hash_map<std::string,std::string>::accessor ac;
988 ac->second = defaultName;
995 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()
static struct flock make_flock(short type, short whence, off_t start, off_t len, pid_t pid)
std::string protocolBufferHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
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)
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_
static void fillDescriptions(edm::ConfigurationDescriptions &descriptions)
ParameterDescriptionBase * addUntracked(U const &iLabel, T const &value)
def create(alignables, pedeDump, additionalData, outputFile, config)
std::string getMergedDatChecksumFilePath(const unsigned int ls, std::string const &stream) const
std::shared_ptr< Json::Value > transferSystemJson_
void setAllowAnything()
allow any parameter label/value pairs
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
void setComment(std::string const &value)
std::mutex * fileDeleteLockPtr_
tbb::concurrent_hash_map< std::string, std::string > mergeTypeMap_
std::string mergeTypePset_
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
std::string getStreamMergeType(std::string const &stream, MergeType defaultType)
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_
void tryInitializeFuLockFile()
void removeFile(unsigned int ls, unsigned int index)
unsigned int stop_ls_override_
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)
std::string stopFilePathPid_
unsigned int eolsNFilesIndex_
std::string getDatFilePath(const unsigned int ls, std::string const &stream) const
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
void openFULockfileStream(bool create)
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)
std::string initFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
unsigned long long uint64_t
void checkMergeTypePSet(edm::ProcessContext const &pc)
ParameterSet const & getParameterSet(std::string const &) const
void preGlobalEndLumi(edm::GlobalContext const &globalContext)
void deserialize(Json::Value &root) override
void watchPostGlobalEndRun(PostGlobalEndRun::slot_type const &iSlot)
LuminosityBlockNumber_t luminosityBlock() const
void add(std::string const &label, ParameterSetDescription const &psetDescription)
std::string & getDefinition()
def remove(d, key, TELL=False)
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
char data[epos_bytes_allocation]
void postEndRun(edm::GlobalContext const &globalContext)
Unserialize a JSON document into a Value.
static const std::vector< std::string > MergeTypeNames_
const_iterator end() const
void checkTransferSystemPSet(edm::ProcessContext const &pc)
std::vector< std::string > const & getNames()
std::string getEoLSFilePathOnFU(const unsigned int ls) const
bool bumpFile(unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize, int maxLS)
void preallocate(edm::service::SystemBounds const &bounds)
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)