16 #include <boost/algorithm/string.hpp>
49 #include <boost/lexical_cast.hpp>
55 defPath_(
pset.getUntrackedParameter<
std::
string>(
"buDefPath",
"")),
56 eventChunkSize_(
pset.getUntrackedParameter<unsigned
int>(
"eventChunkSize", 32) * 1048576),
57 eventChunkBlock_(
pset.getUntrackedParameter<unsigned
int>(
"eventChunkBlock", 32) * 1048576),
58 numBuffers_(
pset.getUntrackedParameter<unsigned
int>(
"numBuffers", 2)),
59 maxBufferedFiles_(
pset.getUntrackedParameter<unsigned
int>(
"maxBufferedFiles", 2)),
60 getLSFromFilename_(
pset.getUntrackedParameter<
bool>(
"getLSFromFilename",
true)),
61 alwaysStartFromFirstLS_(
pset.getUntrackedParameter<
bool>(
"alwaysStartFromFirstLS",
false)),
62 verifyChecksum_(
pset.getUntrackedParameter<
bool>(
"verifyChecksum",
true)),
63 useL1EventID_(
pset.getUntrackedParameter<
bool>(
"useL1EventID",
false)),
67 fileListMode_(
pset.getUntrackedParameter<
bool>(
"fileListMode",
false)),
68 fileListLoopMode_(
pset.getUntrackedParameter<
bool>(
"fileListLoopMode",
false)),
73 currentLumiSection_(0),
74 tcds_pointer_(nullptr),
77 gethostname(thishost, 255);
78 edm::LogInfo(
"FedRawDataInputSource") <<
"Construction. read-ahead chunk size -: " << std::endl
83 throw cms::Exception(
"FedRawDataInputSource::fillFEDRawDataCollection")
84 <<
"Invalid TCDS Test FED range parameter";
90 long autoRunNumber = -1;
94 if (autoRunNumber < 0)
95 throw cms::Exception(
"FedRawDataInputSource::FedRawDataInputSource") <<
"Run number not found from filename";
114 throw cms::Exception(
"FedRawDataInputSource::FedRawDataInputSource")
115 <<
"no reading enabled with numBuffers parameter 0";
122 edm::LogError(
"FedRawDataInputSource::FedRawDataInputSource") <<
"Intel crc32c checksum computation unavailable";
129 edm::LogInfo(
"FedRawDataInputSource") <<
"No FastMonitoringService found in the configuration";
134 throw cms::Exception(
"FedRawDataInputSource") <<
"FastMonitoringService not found";
140 cms::Exception(
"FedRawDataInputSource") <<
"EvFDaqDirector not found";
144 edm::LogInfo(
"FedRawDataInputSource") <<
"EvFDaqDirector/Source configured to use file service";
165 cvReader_.push_back(
new std::condition_variable);
167 threadInit_.store(
false, std::memory_order_release);
187 std::unique_lock<std::mutex> lk(
mReader_);
208 desc.setComment(
"File-based Filter Farm input source for reading raw data from BU ramdisk");
209 desc.addUntracked<
unsigned int>(
"eventChunkSize", 32)->setComment(
"Input buffer (chunk) size");
210 desc.addUntracked<
unsigned int>(
"eventChunkBlock", 32)
211 ->setComment(
"Block size used in a single file read call (must be smaller or equal to buffer size)");
212 desc.addUntracked<
unsigned int>(
"numBuffers", 2)->setComment(
"Number of buffers used for reading input");
213 desc.addUntracked<
unsigned int>(
"maxBufferedFiles", 2)
214 ->setComment(
"Maximum number of simultaneously buffered raw files");
215 desc.addUntracked<
unsigned int>(
"alwaysStartFromfirstLS",
false)
216 ->setComment(
"Force source to start from LS 1 if server provides higher lumisection number");
217 desc.addUntracked<
bool>(
"verifyChecksum",
true)
218 ->setComment(
"Verify event CRC-32C checksum of FRDv5 and higher or Adler32 with v3 and v4");
219 desc.addUntracked<
bool>(
"useL1EventID",
false)
220 ->setComment(
"Use L1 event ID from FED header if true or from TCDS FED if false");
221 desc.addUntracked<std::vector<unsigned int>>(
"testTCDSFEDRange", std::vector<unsigned int>())
222 ->setComment(
"[min, max] range to search for TCDS FED ID in test setup");
223 desc.addUntracked<
bool>(
"fileListMode",
false)
224 ->setComment(
"Use fileNames parameter to directly specify raw files to open");
225 desc.addUntracked<std::vector<std::string>>(
"fileNames", std::vector<std::string>())
226 ->setComment(
"file list used when fileListMode is enabled");
227 desc.setAllowAnything();
228 descriptions.
add(
"source",
desc);
256 open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
267 S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
273 edm::LogInfo(
"FedRawDataInputSource") <<
"----------------RUN ENDED----------------";
316 open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
330 gettimeofday(&tv,
nullptr);
331 const edm::Timestamp lsopentime((
unsigned long long)tv.tv_sec * 1000000 + (
unsigned long long)tv.tv_usec);
339 edm::LogInfo(
"FedRawDataInputSource") <<
"New lumi section was opened. LUMI -: " << lumiSection;
360 std::unique_lock<std::mutex> lkw(
mWakeup_);
371 <<
"Run has been aborted by the input source reader thread";
416 <<
" but according to BU JSON there should be " <<
currentFile_->nEvents_ <<
" events";
420 std::unique_lock<std::mutex> lkw(
mWakeup_);
440 <<
"Premature end of input file while reading file header";
443 <<
"File with only raw header and no events received in LS " <<
currentFile_->lumi_;
460 <<
"Premature end of input file while reading event header";
491 <<
"Premature end of input file while reading event header";
495 event_ = std::make_unique<FRDEventMsgView>(dataPosition);
498 <<
" event id:" <<
event_->event() <<
" lumi:" <<
event_->lumi() <<
" run:" <<
event_->run()
507 <<
"Premature end of input file while reading event data";
513 event_ = std::make_unique<FRDEventMsgView>(dataPosition);
533 unsigned char* dataPosition;
538 event_ = std::make_unique<FRDEventMsgView>(dataPosition);
541 <<
" event id:" <<
event_->event() <<
" lumi:" <<
event_->lumi() <<
" run:" <<
event_->run()
550 <<
"Premature end of input file while reading event data";
573 event_ = std::make_unique<FRDEventMsgView>(dataPosition);
576 chunkEnd =
currentFile_->advance(dataPosition, msgSize);
587 if (crc !=
event_->crc32c()) {
591 <<
"Found a wrong crc32c checksum: expected 0x" << std::hex <<
event_->crc32c() <<
" but calculated 0x"
595 uint32_t adler = adler32(0
L, Z_NULL, 0);
596 adler = adler32(adler, (Bytef*)
event_->payload(),
event_->eventSize());
598 if (adler !=
event_->adler32()) {
602 <<
"Found a wrong Adler32 checksum: expected 0x" << std::hex <<
event_->adler32() <<
" but calculated 0x"
627 <<
"No TCDS or GTP FED in event with FEDHeader EID -: " <<
L1EventID_;
641 static_cast<edm::EventAuxiliary::ExperimentType>(fedHeader.
triggerType()),
668 bool fileIsBeingProcessed =
false;
671 fileIsBeingProcessed =
true;
675 if (!fileIsBeingProcessed) {
692 gettimeofday(&stv,
nullptr);
697 uint32_t eventSize =
event_->eventSize();
698 unsigned char*
event = (
unsigned char*)
event_->payload();
702 uint16_t selectedTCDSFed = 0;
703 while (eventSize > 0) {
713 throw cms::Exception(
"FedRawDataInputSource::fillFEDRawDataCollection") <<
"Out of range FED ID : " <<
fedId;
715 if (!selectedTCDSFed) {
716 selectedTCDSFed =
fedId;
722 throw cms::Exception(
"FedRawDataInputSource::fillFEDRawDataCollection")
723 <<
"Second TCDS FED ID " <<
fedId <<
" found. First ID: " << selectedTCDSFed;
733 tstamp =
edm::Timestamp(static_cast<edm::TimeValue_t>((gpsh << 32) + gpsl));
743 memcpy(fedData.
data(),
event + eventSize, fedSize);
754 unsigned int currentLumiSection = 0;
764 uint32_t lockCount = 0;
774 bool copy_active =
false;
792 std::unique_lock<std::mutex> lkw(
mWakeup_);
797 LogDebug(
"FedRawDataInputSource") <<
"No free chunks or threads...";
813 uint32_t fileSizeIndex;
814 int64_t fileSizeFromMetadata;
822 uint16_t rawHeaderSize = 0;
823 uint32_t lsFromRaw = 0;
824 int32_t serverEventsInNewFile = -1;
847 serverEventsInNewFile,
848 fileSizeFromMetadata,
869 serverEventsInNewFile,
870 fileSizeFromMetadata,
881 if (thisLockWaitTimeUs > 0.)
882 sumLockWaitTimeUs += thisLockWaitTimeUs;
890 sumLockWaitTimeUs = 0;
920 if (
ls > currentLumiSection) {
924 currentLumiSection =
ls;
934 for (
unsigned int nextLS = lsToStart; nextLS <=
ls; nextLS++) {
945 for (
unsigned int nextLS = currentLumiSection + 1; nextLS <=
ls; nextLS++) {
950 currentLumiSection =
ls;
954 if (currentLumiSection > 0 &&
ls < currentLumiSection) {
956 <<
"Got old LS (" <<
ls <<
") file from EvFDAQDirector! Expected LS:" << currentLumiSection
957 <<
". Aborting execution." << std::endl;
972 if (!(dbgcount % 20))
973 LogDebug(
"FedRawDataInputSource") <<
"No file for me... sleep and try again...";
977 backoff_exp =
std::min(4, backoff_exp);
979 int sleeptime = (
int)(100000. *
pow(2, backoff_exp));
989 LogDebug(
"FedRawDataInputSource") <<
"The director says to grab -: " << nextFile;
997 rawFile = rawFilePath.replace_extension(
".raw").string();
1001 int stat_res =
stat(rawFile.c_str(), &st);
1002 if (stat_res == -1) {
1003 edm::LogError(
"FedRawDataInputSource") <<
"Can not stat file (" << errno <<
"):-" << rawFile << std::endl;
1014 int eventsInNewFile;
1017 eventsInNewFile = 0;
1019 eventsInNewFile = -1;
1023 if (rawHeaderSize) {
1024 int rawFdEmpty = -1;
1025 uint16_t rawHeaderCheck;
1028 nextFile, rawFdEmpty, rawHeaderCheck, fileSizeFromMetadata, fileFound, 0,
true);
1029 assert(fileFound && rawHeaderCheck == rawHeaderSize);
1034 eventsInNewFile = serverEventsInNewFile;
1035 assert(eventsInNewFile >= 0);
1036 assert((eventsInNewFile > 0) ==
1037 (fileSize > rawHeaderSize));
1046 std::unique_ptr<InputFile> newInputFile(
new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1057 auto newInputFilePtr = newInputFile.get();
1060 for (
unsigned int i = 0;
i < neededChunks;
i++) {
1062 bool copy_active =
false;
1072 unsigned int newTid = 0xffffffff;
1082 bool copy_active =
false;
1100 if (newChunk ==
nullptr) {
1102 if (newTid != 0xffffffff)
1111 std::unique_lock<std::mutex> lk(
mReader_);
1125 if (!eventsInNewFile) {
1131 std::unique_lock<std::mutex> lkw(
mWakeup_);
1133 std::unique_ptr<InputFile> newInputFile(
new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1140 (rawHeaderSize > 0),
1157 std::unique_lock<std::mutex> lkw(
mWakeup_);
1162 newChunk->
reset(0, toRead, 0);
1166 std::unique_ptr<InputFile> newInputFile(
new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1176 newInputFile->chunks_[0] = newChunk;
1185 unsigned numFinishedThreads = 0;
1191 std::unique_lock<std::mutex> lk(
mReader_);
1194 numFinishedThreads++;
1204 threadInit_.exchange(
true, std::memory_order_acquire);
1208 std::unique_lock<std::mutex> lk(
mReader_);
1235 unsigned int bufferLeft = (chunk->
offset_ == 0 &&
file->rawFd_ != -1) ?
file->rawHeaderSize_ : 0;
1240 bool fileOpenedHere =
false;
1243 fileDescriptor =
file->rawFd_;
1244 if (fileDescriptor == -1) {
1245 fileDescriptor = open(
file->fileName_.c_str(), O_RDONLY);
1246 fileOpenedHere =
true;
1247 file->rawFd_ = fileDescriptor;
1251 fileDescriptor =
file->rawFd_;
1253 if (fileDescriptor == -1) {
1254 fileDescriptor = open(
file->fileName_.c_str(), O_RDONLY);
1255 fileOpenedHere =
true;
1258 fileDescriptor = open(
file->fileName_.c_str(), O_RDONLY);
1259 fileOpenedHere =
true;
1263 if (fileDescriptor < 0) {
1264 edm::LogError(
"FedRawDataInputSource") <<
"readWorker failed to open file -: " <<
file->fileName_
1265 <<
" fd:" << fileDescriptor <<
" error: " << strerror(errno);
1269 if (fileOpenedHere) {
1271 pos = lseek(fileDescriptor, chunk->
offset_, SEEK_SET);
1274 <<
"readWorker failed to seek file -: " <<
file->fileName_ <<
" fd:" << fileDescriptor <<
" to offset "
1275 << chunk->
offset_ <<
" error: " << strerror(errno);
1281 LogDebug(
"FedRawDataInputSource") <<
"Reader thread opened file -: TID: " << tid <<
" file: " <<
file->fileName_
1282 <<
" at offset " << lseek(fileDescriptor, 0, SEEK_CUR);
1284 unsigned int skipped = bufferLeft;
1294 edm::LogError(
"FedRawDataInputSource") <<
"readWorker failed to read file -: " <<
file->fileName_
1295 <<
" fd:" << fileDescriptor <<
" error: " << strerror(errno);
1305 <<
"readWorker failed to read file -: " <<
file->fileName_ <<
" fd:" << fileDescriptor <<
" last:" <<
last
1306 <<
" expectedChunkSize:" << chunk->
usedSize_
1308 <<
" block:" << (
i + 1) <<
"/" <<
readBlocks_ <<
" error: " << strerror(errno);
1319 std::chrono::milliseconds msec = std::chrono::duration_cast<std::chrono::milliseconds>(
diff);
1320 LogDebug(
"FedRawDataInputSource") <<
" finished reading block -: " << (bufferLeft >> 20) <<
" MB"
1321 <<
" in " << msec.count() <<
" ms (" << (bufferLeft >> 20) /
double(msec.count())
1324 if (chunk->
offset_ + bufferLeft ==
file->fileSize_) {
1325 close(fileDescriptor);
1326 fileDescriptor = -1;
1331 close(fileDescriptor);
1346 throw cms::Exception(
"FedRawDataInputSource:threadError") <<
" file reader thread error ";
1373 if (currentLeft <
size) {
1421 LogDebug(
"FedRawDataInputSource:InputFile") <<
"Deleting input file -:" <<
fileName_;
1424 }
catch (
const std::filesystem::filesystem_error& ex) {
1426 <<
" - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() <<
". Trying again.";
1429 <<
" - deleteFile std::exception CAUGHT -: " << ex.what() <<
". Trying again.";
1437 uint32_t existingSize = 0;
1441 if (
file->rawFd_ == -1) {
1443 if (
file->rawHeaderSize_)
1450 existingSize +=
file->rawHeaderSize_;
1453 LogDebug(
"FedRawDataInputSource") <<
"opened file -: " << std::endl <<
file->fileName_;
1455 throw cms::Exception(
"FedRawDataInputSource:readNextChunkIntoBuffer")
1456 <<
"failed to open file " << std::endl
1462 (
void*)(
file->chunks_[0]->buf_ + existingSize),
1465 existingSize +=
last;
1470 if (
file->chunkPosition_ == 0) {
1474 existingSize +=
last;
1479 memmove((
void*)
file->chunks_[0]->buf_,
file->chunks_[0]->buf_ +
file->chunkPosition_, existingSizeLeft);
1485 for (uint32_t
i = 0;
i < blockcount;
i++) {
1486 const ssize_t
last =
1489 existingSizeLeft +=
last;
1495 file->chunkPosition_ = 0;
1500 LogDebug(
"FedRawDataInputSource") <<
"Closing input file -: " << std::endl <<
file->fileName_;
1520 std::pair<bool, unsigned int>
ret(
true, itr->second);
1525 return std::pair<bool, unsigned int>(
false, 0);
1530 if (
a.rfind(
'/') != std::string::npos)
1531 a =
a.substr(
a.rfind(
'/'));
1532 if (
b.rfind(
'/') != std::string::npos)
1533 b =
b.substr(
b.rfind(
'/'));
1541 if (fileStem.find(
"file://") == 0)
1542 fileStem = fileStem.substr(7);
1543 else if (fileStem.find(
"file:") == 0)
1544 fileStem = fileStem.substr(5);
1545 auto end = fileStem.find(
'_');
1547 if (fileStem.find(
"run") == 0) {
1551 long rval = boost::lexical_cast<long>(runStr);
1552 edm::LogInfo(
"FedRawDataInputSource") <<
"Autodetected run number in fileListMode -: " << rval;
1554 }
catch (boost::bad_lexical_cast
const&) {
1556 <<
"Unable to autodetect run number in fileListMode from file -: " <<
fileName;
1569 if (nextFile.find(
"file://") == 0)
1570 nextFile = nextFile.substr(7);
1571 else if (nextFile.find(
"file:") == 0)
1572 nextFile = nextFile.substr(5);
1575 if (fileStem.find(
"ls"))
1576 fileStem = fileStem.substr(fileStem.find(
"ls") + 2);
1577 if (fileStem.find(
'_'))
1578 fileStem = fileStem.substr(0, fileStem.find(
'_'));
1581 ls = boost::lexical_cast<unsigned int>(fileStem);
1596 return getFile(
ls, nextFile, fsize, lockWaitTime);