From 2c574a4c6b3cdb49b2cd01ec156be1480c98ddb2 Mon Sep 17 00:00:00 2001 From: shahoian Date: Wed, 6 May 2026 18:52:27 +0200 Subject: [PATCH] Raw TF dump workflow --- Common/Utils/include/CommonUtils/NameConf.h | 3 + Common/Utils/src/NameConf.cxx | 5 + Detectors/Raw/TFReaderDD/CMakeLists.txt | 7 + .../Raw/TFReaderDD/src/RawTFDumpSpec.cxx | 460 ++++++++++++++++++ Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.h | 23 + .../TFReaderDD/src/SubTimeFrameFileReader.cxx | 5 +- .../TFReaderDD/src/tf-data-dump-workflow.cxx | 43 ++ 7 files changed, 544 insertions(+), 2 deletions(-) create mode 100644 Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.cxx create mode 100644 Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.h create mode 100644 Detectors/Raw/TFReaderDD/src/tf-data-dump-workflow.cxx diff --git a/Common/Utils/include/CommonUtils/NameConf.h b/Common/Utils/include/CommonUtils/NameConf.h index fb10f929c9782..8d4c0a2c1c4f8 100644 --- a/Common/Utils/include/CommonUtils/NameConf.h +++ b/Common/Utils/include/CommonUtils/NameConf.h @@ -100,6 +100,9 @@ class NameConf : public o2::conf::ConfigurableParamHelper // CTF Dictionary static std::string getCTFDictFileName(); + // O2 Raw TF Filename + static std::string getRawTFFileName(uint32_t run, uint32_t orb, uint32_t id, const std::string& host, const std::string_view prefix = "o2_rawtf_dump"); + // Default CCDB server static std::string getCCDBServer(); diff --git a/Common/Utils/src/NameConf.cxx b/Common/Utils/src/NameConf.cxx index 45646284a878b..48cefacaf14c7 100644 --- a/Common/Utils/src/NameConf.cxx +++ b/Common/Utils/src/NameConf.cxx @@ -95,6 +95,11 @@ std::string NameConf::getCTFFileName(uint32_t run, uint32_t orb, uint32_t id, co return o2::utils::Str::concat_string(prefix, '_', fmt::format("run{:08d}_orbit{:010d}_tf{:010d}_{}", run, orb, id, host), ".root"); } +std::string NameConf::getRawTFFileName(uint32_t run, uint32_t orb, uint32_t id, const std::string& host, const std::string_view prefix) +{ + return o2::utils::Str::concat_string(prefix, '_', fmt::format("run{:08d}_orbit{:010d}_tf{:010d}_{}", run, orb, id, host), ".tf"); +} + std::string NameConf::getCTFDictFileName() { return o2::utils::Str::concat_string(CTFDICT, ".root"); diff --git a/Detectors/Raw/TFReaderDD/CMakeLists.txt b/Detectors/Raw/TFReaderDD/CMakeLists.txt index 12ecc9ca8795d..f87d1b5a7704e 100644 --- a/Detectors/Raw/TFReaderDD/CMakeLists.txt +++ b/Detectors/Raw/TFReaderDD/CMakeLists.txt @@ -26,3 +26,10 @@ o2_add_executable(tf-reader-workflow SOURCES src/TFReaderSpec.cxx src/tf-reader-workflow.cxx PUBLIC_LINK_LIBRARIES O2::TFReaderDD) + + +o2_add_executable(tf-dump-workflow + COMPONENT_NAME raw + SOURCES src/RawTFDumpSpec.cxx + src/tf-data-dump-workflow.cxx + PUBLIC_LINK_LIBRARIES O2::TFReaderDD) diff --git a/Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.cxx b/Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.cxx new file mode 100644 index 0000000000000..2baadf8bdebfd --- /dev/null +++ b/Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.cxx @@ -0,0 +1,460 @@ +// Copyright 2019-2020 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +#include "Framework/WorkflowSpec.h" +#include "Framework/ConfigParamRegistry.h" +#include "Framework/RawDeviceService.h" +#include "Framework/InputRecordWalker.h" +#include "Framework/Task.h" +#include "Framework/DataTakingContext.h" +#include "Framework/TimingInfo.h" +#include "DataFormatsParameters/GRPECSObject.h" +#include "DetectorsCommonDataFormats/FileMetaData.h" +#include "RawTFDumpSpec.h" +#include "TFReaderDD/SubTimeFrameFile.h" +#include "CommonUtils/NameConf.h" +#include "CommonUtils/FileSystemUtils.h" +#include "CommonUtils/StringUtils.h" +#include "Algorithm/RangeTokenizer.h" +#include + +namespace o2::rawdd +{ +namespace o2h = o2::header; +using namespace o2::framework; +using DataHeader = o2::header::DataHeader; +using DetID = o2::detectors::DetID; +using ios = std::ios_base; + +class RawTFDump : public Task +{ + public: + static constexpr o2h::DataDescription DESCRaw{"RAWDATA"}, DESCCRaw{"CRAWDATA"}; + + RawTFDump(bool); + void init(InitContext& ic) final; + void run(ProcessingContext& pc) final; + void endOfStream(EndOfStreamContext& ec) final; + + private: + void updateTimeDependentParams(ProcessingContext& pc); + void prepareTFForWriting(ProcessingContext& pc); + size_t getTFSizeInFile() const; + size_t getCurrentFileSize(); + void prepareTFFile(); + void closeTFFile(); + bool checkFreeSpace(ProcessingContext& pc); + + SubTimeFrameFileDataIndex mTFDataIndex; + std::vector> mTFData; + std::map> mDataMap; + std::vector mFilter{}; + + size_t mTFSize = 0; + + size_t mMinSize = 0; // if > 0, accumulate TFs in the same file until the total size exceeds this minimum + size_t mMaxSize = 0; // if > MinSize, and accumulated size will exceed this value, stop accumulation (even if mMinSize is not reached) + size_t mNTFs = 0; // total number of TFs written + size_t mNAccTF = 0; // total number of TFs accumulated in the current file + size_t mNTFFiles = 0; // total number of TF files written + int mMaxTFPerFile = 0; // max TFs per files to store + int mWaitDiskFull = 0; // if mCheckDiskFull triggers, pause for this amount of ms before new attempt + int mWaitDiskFullMax = -1; // produce fatal mCheckDiskFull block the workflow for more than this time (in ms) + float mCheckDiskFull = 0.; // wait for if available abs. disk space is < mCheckDiskFull (if >0) or if its fraction is < -mCheckDiskFull (if <0) + float mMaxAccRate = 0.f; // max acceptance rate + bool mFillMD5 = false; + bool mWriteTF = true; // for dry run + bool mStoreMetaFile = false; + bool mCreateRunEnvDir = true; + bool mRejectCurrentTF = false; + bool mRejectDEADBEEF = false; + bool mVerbose = false; + std::vector mTFOrbits{}; // 1st orbits of TF accumulated in current file + o2::framework::DataTakingContext mDataTakingContext{}; + o2::framework::TimingInfo mTimingInfo{}; + + std::string mHostName{}; + std::string mTFDir{}; + std::string mTFMetaFileDir = "/dev/null"; + std::string mCurrentTFFileName{}; + std::string mCurrentTFFileNameFull{}; + std::string mCurrentTFFileNameFullTmp{}; + std::string mMetaDataType{}; + + static constexpr size_t MiB = 1ul << 20; + static constexpr std::streamsize sBuffSize = MiB; // 1 MiB + static constexpr std::streamsize sChunkSize = 512; + static const std::string TMPFileEnding; + std::unique_ptr mFileBuf; + std::ofstream mFile; + + // helper to make sure the written blocks are buffered + template < + typename pointer, + typename std::enable_if< + std::is_pointer::value && // pointers only + (std::is_void>::value || // void* or standard layout! + std::is_standard_layout>::value)>::type* = nullptr> + void buffered_write(const pointer p, std::streamsize pCount) + { + // make sure we're not doing a short write + assert((pCount % sizeof(std::conditional_t>::value, + char, std::remove_pointer_t>) == + 0) && + "Performing short write?"); + + const char* lPtr = reinterpret_cast(p); + // avoid the optimization if the write is large enough + if (pCount >= sBuffSize) { + mFile.write(lPtr, pCount); + } else { + // split the write to smaller chunks + while (pCount > 0) { + const auto lToWrite = std::min(pCount, sChunkSize); + assert(lToWrite > 0 && lToWrite <= sChunkSize && lToWrite <= pCount); + + mFile.write(lPtr, lToWrite); + lPtr += lToWrite; + pCount -= lToWrite; + } + } + } +}; + +const std::string RawTFDump::TMPFileEnding{".part"}; + +//________________________________________ +RawTFDump::RawTFDump(bool) +{ + mFileBuf = std::make_unique(sBuffSize); + mFile.rdbuf()->pubsetbuf(mFileBuf.get(), sBuffSize); + mFile.clear(); + mFile.exceptions(std::fstream::failbit | std::fstream::badbit); +} + +//________________________________________ +void RawTFDump::init(InitContext& ic) +{ + + mTFMetaFileDir = ic.options().get("meta-output-dir"); + if (mTFMetaFileDir != "/dev/null") { + mTFMetaFileDir = o2::utils::Str::rectifyDirectory(mTFMetaFileDir); + mStoreMetaFile = true; + mFillMD5 = ic.options().get("md5-for-meta"); + } + + mTFDir = ic.options().get("output-dir"); + if (mTFDir != "/dev/null") { + mTFDir = o2::utils::Str::rectifyDirectory(mTFDir); + mWriteTF = true; + } else { + mWriteTF = false; + mStoreMetaFile = false; + } + + mRejectDEADBEEF = !ic.options().get("include-deadbeef"); + mCreateRunEnvDir = !ic.options().get("ignore-partition-run-dir"); + mMinSize = ic.options().get("min-file-size"); + mMaxSize = ic.options().get("max-file-size"); + mMaxTFPerFile = ic.options().get("max-tf-per-file"); + mMaxAccRate = ic.options().get("max-tf-write-rate"); + mVerbose = ic.options().get("use-verbose-mode"); + if (mMaxAccRate > 0) { // RSTODO + LOGP(info, "Will reject{} {}% of TFs", mMaxAccRate < 100 ? " randomly" : "", mMaxAccRate < 100 ? mMaxAccRate : 100); + } else if (mMaxAccRate < -1) { + LOGP(info, "Will reject all but each {}-th TF slice", -mMaxAccRate); + } + + if (mWriteTF) { + if (mMinSize > 0) { + LOGP(info, "Multiple TFs will be accumulated in the file until its size exceeds {}{}", + mMinSize, mMaxSize > mMinSize ? fmt::format(" but does not exceed {} B", mMaxSize) : std::string{}); + } + } + + mCheckDiskFull = ic.options().get("require-free-disk"); + mWaitDiskFull = 1000 * ic.options().get("wait-for-free-disk"); + mWaitDiskFullMax = 1000 * ic.options().get("max-wait-for-free-disk"); + + char hostname[_POSIX_HOST_NAME_MAX]; + gethostname(hostname, _POSIX_HOST_NAME_MAX); + mHostName = hostname; + mHostName = mHostName.substr(0, mHostName.find('.')); +} + +//________________________________________ +size_t RawTFDump::getTFSizeInFile() const +{ + return SubTimeFrameFileMeta::getSizeInFile() + mTFDataIndex.getSizeInFile() + mTFSize; +} + +//________________________________________ +size_t RawTFDump::getCurrentFileSize() +{ + return mFile.is_open() ? size_t(mFile.tellp()) : 0; +} + +//___________________________________________________________________ +void RawTFDump::prepareTFFile() +{ + if (!mWriteTF) { + return; + } + bool needToOpen; + if (!mFile.is_open()) { + needToOpen = true; + } else { + auto currSize = getCurrentFileSize(); + if ((mNAccTF >= mMaxTFPerFile) || + (currSize >= mMinSize) || // min size exceeded, may close the file. + (currSize && mMaxSize > mMinSize && ((currSize + mTFSize) > mMaxSize))) { // this is not the 1st TF in the file and the new size will exceed allowed max + needToOpen = true; + } else { + LOGP(info, "Will add new TF of size {} to existing file of size {} with {} TFs", mTFSize, currSize, mNAccTF); + needToOpen = false; + } + } + if (needToOpen) { + closeTFFile(); + auto TFDir = mTFDir.empty() ? o2::utils::Str::rectifyDirectory("./") : mTFDir; + if (mCreateRunEnvDir && !mDataTakingContext.envId.empty() && (mDataTakingContext.envId != o2::framework::DataTakingContext::UNKNOWN)) { + TFDir += fmt::format("{}_{}tf/", mDataTakingContext.envId, mDataTakingContext.runNumber); + if (!TFDir.empty()) { + o2::utils::createDirectoriesIfAbsent(TFDir); + LOGP(info, "Created {} directory for TFs output", TFDir); + } + } + mCurrentTFFileName = o2::base::NameConf::getRawTFFileName(mTimingInfo.runNumber, mTimingInfo.firstTForbit, mTimingInfo.tfCounter, mHostName); + mCurrentTFFileNameFull = fmt::format("{}{}", TFDir, mCurrentTFFileName); + mCurrentTFFileNameFullTmp = TMPFileEnding.empty() ? mCurrentTFFileNameFull : o2::utils::Str::concat_string(mCurrentTFFileNameFull, TMPFileEnding); + mFile.open(mCurrentTFFileNameFullTmp.c_str(), ios::binary | ios::trunc | ios::out | ios::ate); + LOGP(info, "Opened new raw-tf dump file {}[{}]", mCurrentTFFileNameFull, TMPFileEnding); + mNTFFiles++; + } +} + +//___________________________________________________________________ +void RawTFDump::updateTimeDependentParams(ProcessingContext& pc) +{ + namespace GRPECS = o2::parameters::GRPECS; + mTimingInfo = pc.services().get(); + if (mTimingInfo.globalRunNumberChanged) { + mDataTakingContext = pc.services().get(); + // determine the output type for the TF metadata + mMetaDataType = GRPECS::getRawDataPersistencyMode(mDataTakingContext.runType, mDataTakingContext.forcedRaw); + } +} + +//___________________________________________________________________ +void RawTFDump::closeTFFile() +{ + if (!mFile.is_open()) { + return; + } + try { + LOGP(info, "Closing output file {}[{}]", mCurrentTFFileNameFull, TMPFileEnding); + mFile.close(); + // write TF file metaFile data + if (mStoreMetaFile) { + o2::dataformats::FileMetaData TFMetaData; + if (!TFMetaData.fillFileData(mCurrentTFFileNameFullTmp, mFillMD5, TMPFileEnding)) { + throw std::runtime_error("metadata file was requested but not created"); + } + TFMetaData.setDataTakingContext(mDataTakingContext); + TFMetaData.type = mMetaDataType; + TFMetaData.priority = "high"; + TFMetaData.tfOrbits.swap(mTFOrbits); + auto metaFileNameTmp = fmt::format("{}{}.tmp", mTFMetaFileDir, mCurrentTFFileName); + auto metaFileName = fmt::format("{}{}.done", mTFMetaFileDir, mCurrentTFFileName); + try { + std::ofstream metaFileOut(metaFileNameTmp); + metaFileOut << TFMetaData; + metaFileOut.close(); + if (!TMPFileEnding.empty()) { + std::filesystem::rename(mCurrentTFFileNameFullTmp, mCurrentTFFileNameFull); + } + std::filesystem::rename(metaFileNameTmp, metaFileName); + LOGP(info, "wrote meta file {}", metaFileName); + } catch (std::exception const& e) { + LOGP(error, "Failed to store TF meta data file {}, reason {}", metaFileName, e.what()); + } + } else if (!TMPFileEnding.empty()) { + std::filesystem::rename(mCurrentTFFileNameFullTmp, mCurrentTFFileNameFull); + } + } catch (std::exception const& e) { + LOGP(error, "Failed to finalize TF file {}, reason: ", mCurrentTFFileNameFull, e.what()); + } + mTFOrbits.clear(); + mNAccTF = 0; +} + +//________________________________________ +void RawTFDump::run(ProcessingContext& pc) +{ + updateTimeDependentParams(pc); + + if (!mRejectCurrentTF) { + prepareTFForWriting(pc); + } + + prepareTFFile(); + if (mWriteTF && checkFreeSpace(pc)) { // write data + try { + size_t lTFSizeInFile = getTFSizeInFile(); + SubTimeFrameFileMeta lTFFileMeta(lTFSizeInFile); + + mFile << lTFFileMeta; // Write DataHeader + SubTimeFrameFileMeta + mFile << mTFDataIndex; // Write DataHeader + SubTimeFrameFileDataIndex + + for (const auto& eqEntry : mDataMap) { + auto& [lSize, lCnt, lEntry] = eqEntry.second; + for (size_t part = 0; part < lCnt; part++) { + const auto& dataPtr = mTFData[lEntry + part]; + DataHeader hdToWrite = *reinterpret_cast(dataPtr.first); // make a local DataHeader copy to clear flagsNextHeader bit + hdToWrite.flagsNextHeader = 0; + buffered_write(reinterpret_cast(&hdToWrite), sizeof(DataHeader)); + buffered_write(dataPtr.second, hdToWrite.payloadSize); + } + } + mFile.flush(); // flush the buffer and check the state + mTFOrbits.push_back(mTimingInfo.firstTForbit); + mNAccTF++; + } catch (const std::ios_base::failure& eFailExc) { + LOGP(error, "Writing of TF {} to file {} failed. error={}", mTimingInfo.tfCounter, mCurrentTFFileNameFullTmp, eFailExc.what()); + } + } + // cleanup + mTFData.clear(); + mDataMap.clear(); + mTFDataIndex.clear(); + mTFSize = 0; +} + +//________________________________________ +bool RawTFDump::checkFreeSpace(ProcessingContext& pc) +{ + int totalWait = 0, nwaitCycles = 0; + while (mCheckDiskFull) { + constexpr int showFirstN = 10, prsecaleWarnings = 50; + try { + const auto si = std::filesystem::space(mCurrentTFFileNameFullTmp); + std::string wmsg{}; + if (mCheckDiskFull > 0.f && si.available < mCheckDiskFull) { + nwaitCycles++; + wmsg = fmt::format("Disk has {} MiB available while at least {} MiB is requested, wait for {} ms (on top of {} ms)", si.available / MiB, size_t(mCheckDiskFull) / MiB, mWaitDiskFull, totalWait); + } else if (mCheckDiskFull < 0.f && float(si.available) / si.capacity < -mCheckDiskFull) { // relative margin requested + nwaitCycles++; + wmsg = fmt::format("Disk has {:.3f}% available while at least {:.3f}% is requested, wait for {} ms (on top of {} ms)", si.capacity ? float(si.available) / si.capacity * 100.f : 0., -mCheckDiskFull, mWaitDiskFull, totalWait); + } else { + nwaitCycles = 0; + } + if (nwaitCycles) { + if (mWaitDiskFullMax > 0 && totalWait > mWaitDiskFullMax) { + closeTFFile(); // try to save whatever we have + LOGP(fatal, "Disk has {} MiB available out of {} MiB after waiting for {} ms", si.available / MiB, si.capacity / MiB, mWaitDiskFullMax); + } + if (nwaitCycles < showFirstN + 1 || (prsecaleWarnings && (nwaitCycles % prsecaleWarnings) == 0)) { + LOGP(alarm, "{}", wmsg); + } + pc.services().get().waitFor((unsigned int)(mWaitDiskFull)); + totalWait += mWaitDiskFull; + continue; + } + } catch (std::exception const& e) { + LOGP(fatal, "unable to query disk space info for path {}, reason {}", mCurrentTFFileNameFull, e.what()); // do we want this? + } + break; + } + return true; +} + +//________________________________________ +void RawTFDump::prepareTFForWriting(ProcessingContext& pc) +{ + for (auto const& ref : InputRecordWalker(pc.inputs(), mFilter)) { + auto const* dh = DataRefUtils::getHeader(ref); + if (!dh) { + LOGP(error, "Failed to extract header"); + continue; + } + if (dh->subSpecification == 0xdeadbeef && mRejectDEADBEEF) { + continue; + } + const auto lHdrDataSize = sizeof(DataHeader) + dh->payloadSize; + mTFSize += lHdrDataSize; + + auto& [lSize, lCnt, lEntry] = mDataMap[EquipmentIdentifier(*dh)]; + if (!lCnt) { + lEntry = mTFData.size(); // flag where the data of this spec starts + } + lSize += lHdrDataSize; + lCnt++; + mTFData.push_back({ref.header, ref.payload}); + if (mVerbose) { + LOGP(info, "{}, part: {} of {}, payload {}, 1stTFOrbit: {} TF: {}", + DataSpecUtils::describe(OutputSpec{dh->dataOrigin, dh->dataDescription, dh->subSpecification}), + dh->splitPayloadIndex, dh->splitPayloadParts, dh->payloadSize, dh->firstTForbit, dh->tfCounter); + } + } + + // build the index + { + LOGP(info, "Creating dump image for TF {} of run {}, starting orbit {}, size = {}", mTimingInfo.tfCounter, mTimingInfo.runNumber, mTimingInfo.firstTForbit, mTFSize); + std::uint64_t lCurrOff = 0; + for (const auto& eqEntry : mDataMap) { + const auto& eq = eqEntry.first; + auto& [lSize, lCnt, lEntry] = eqEntry.second; + assert(lSize > sizeof(DataHeader)); + + OutputSpec spec{eq.mDataOrigin, eq.mDataDescription, eq.mSubSpecification}; + if (mVerbose) { + LOGP(info, "{} : {} parts of size {} | offset: {}", DataSpecUtils::describe(spec), lCnt, lSize, lCurrOff); + } + mTFDataIndex.AddStfElement(eq, lCnt, lCurrOff, lSize); + lCurrOff += lSize; + } + } + mNTFs++; +} + +//____________________________________________________________ +void RawTFDump::endOfStream(EndOfStreamContext&) +{ + closeTFFile(); + LOGP(info, "Dumped {} TFs in {} files", mNTFs, mNTFFiles); +} + +//__________________________________________________________ +DataProcessorSpec getRawTFDumpSpec(const std::string& inpconfig) +{ + std::vector inputs = select(inpconfig.c_str()); + return DataProcessorSpec{ + "raw-tf-dump", + inputs, + {}, + AlgorithmSpec{adaptFromTask(false)}, + Options{ + {"include-deadbeef", VariantType::Bool, false, {"Include DPL-generated 0xdeadbeef subspecs for missing data"}}, + {"output-dir", VariantType::String, "none", {"TF output directory, must exist"}}, + {"meta-output-dir", VariantType::String, "/dev/null", {"TF metadata output directory, must exist (if not /dev/null)"}}, + {"md5-for-meta", VariantType::Bool, false, {"fill CTF file MD5 sum in the metadata file"}}, + {"min-file-size", VariantType::Int64, 0l, {"accumulate TFs until given file size reached"}}, + {"max-file-size", VariantType::Int64, 0l, {"if > 0, try to avoid exceeding given file size, also used for space check"}}, + {"max-tf-per-file", VariantType::Int, 0, {"if > 0, avoid storing more than requested CTFs per file"}}, + {"max-tf-write-rate", VariantType::Int, 0, {">0: percentage to reject randomly, <0: reject if timeslice%|value|!=0"}}, + {"require-free-disk", VariantType::Float, 0.f, {"pause writing op. if available disk space is below this margin, in bytes if >0, as a fraction of total if <0"}}, + {"wait-for-free-disk", VariantType::Float, 10.f, {"if paused due to the low disk space, recheck after this time (in s)"}}, + {"max-wait-for-free-disk", VariantType::Float, 60.f, {"produce fatal if paused due to the low disk space for more than this amount in s."}}, + {"use-verbose-mode", VariantType::Bool, false, {"Use verbose mode"}}, + {"ignore-partition-run-dir", VariantType::Bool, false, {"Do not creare partition-run directory in output-dir"}}}}; +} + +} // namespace o2::rawdd diff --git a/Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.h b/Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.h new file mode 100644 index 0000000000000..ce4c33522c7a2 --- /dev/null +++ b/Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.h @@ -0,0 +1,23 @@ +// Copyright 2019-2020 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +#ifndef O2_RAW_TF_DUMP_SPEC_ +#define O2_RAW_TF_DUMP_SPEC_ + +#include "DetectorsCommonDataFormats/DetID.h" +#include "Framework/DeviceSpec.h" + +namespace o2::rawdd +{ +o2::framework::DataProcessorSpec getRawTFDumpSpec(const std::string& inpconfig); +} + +#endif diff --git a/Detectors/Raw/TFReaderDD/src/SubTimeFrameFileReader.cxx b/Detectors/Raw/TFReaderDD/src/SubTimeFrameFileReader.cxx index f227390e67ef3..5f862dffe512f 100644 --- a/Detectors/Raw/TFReaderDD/src/SubTimeFrameFileReader.cxx +++ b/Detectors/Raw/TFReaderDD/src/SubTimeFrameFileReader.cxx @@ -115,7 +115,7 @@ std::size_t SubTimeFrameFileReader::getHeaderStackSize() // throws ios_base::fai LOGP(error, "FileReader: Reached max number of headers allowed: {}.", cMaxHeaders); return 0; } - + LOGP(debug, "getHeaderStackSize, pos = {}, size = {}", lFilePosStart, lStackSize); return lStackSize; } @@ -252,6 +252,7 @@ std::unique_ptr SubTimeFrameFileReader::read(fair::mq::Device* return nullptr; } lStfMetaDataHdr = o2::header::DataHeader::Get(lMetaHdrStack.first()); + LOGP(debug, "read filemeta, pos = {}, size = {}", position(), sizeof(SubTimeFrameFileMeta)); if (!read_advance(&lStfFileMeta, sizeof(SubTimeFrameFileMeta))) { return nullptr; } @@ -320,7 +321,6 @@ std::unique_ptr SubTimeFrameFileReader::read(fair::mq::Device* STFHeader stfHeader{tfID, -1u, -1u}; // read pairs while (lLeftToRead > 0) { - // allocate and read the Headers std::size_t lDataHeaderStackSize = 0; Stack lDataHeaderStack = getHeaderStack(lDataHeaderStackSize); @@ -398,6 +398,7 @@ std::unique_ptr SubTimeFrameFileReader::read(fair::mq::Device* msgSW.Stop(); #endif memcpy(lHdrStackMsg->GetData(), headerStack.data(), headerStack.size()); + LOGP(debug, "read data, pos = {}, size = {} leftToRead {}", position(), lDataSize, lLeftToRead); if (!read_advance(lDataMsg->GetData(), lDataSize)) { return nullptr; diff --git a/Detectors/Raw/TFReaderDD/src/tf-data-dump-workflow.cxx b/Detectors/Raw/TFReaderDD/src/tf-data-dump-workflow.cxx new file mode 100644 index 0000000000000..63ec54392b869 --- /dev/null +++ b/Detectors/Raw/TFReaderDD/src/tf-data-dump-workflow.cxx @@ -0,0 +1,43 @@ +// Copyright 2019-2020 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +#include "CommonUtils/ConfigurableParam.h" +#include "Framework/ConfigParamRegistry.h" +#include "Framework/CompletionPolicy.h" +#include "Framework/CompletionPolicyHelpers.h" + +using namespace o2::framework; + +void customize(std::vector& workflowOptions) +{ + std::vector options; + options.push_back(ConfigParamSpec{"dataspec", VariantType::String, "tst:TST/A", {"selection string for the data to be proxied"}}); + options.push_back(ConfigParamSpec{"configKeyValues", VariantType::String, "", {"semicolon separated key=value strings"}}); + std::swap(workflowOptions, options); +} + +void customize(std::vector& policies) +{ + policies.push_back({CompletionPolicyHelpers::consumeWhenPastOldestPossibleTimeframe("raw-tf-dump", [](auto const&) -> bool { return true; })}); +} + +// ------------------------------------------------------------------ + +#include "Framework/runDataProcessing.h" +#include "RawTFDumpSpec.h" + +WorkflowSpec defineDataProcessing(ConfigContext const& configcontext) +{ + o2::conf::ConfigurableParam::updateFromString(configcontext.options().get("configKeyValues")); + auto inpconfig = configcontext.options().get("dataspec"); + WorkflowSpec specs{o2::rawdd::getRawTFDumpSpec(inpconfig)}; + return specs; +}