diff --git a/algo/CMakeLists.txt b/algo/CMakeLists.txt
index 7992e1d36aa2f969fce915b7a9cdb518a0974bf7..bc7874f883f858201230207ffcc8adc3eb4bbe25 100644
--- a/algo/CMakeLists.txt
+++ b/algo/CMakeLists.txt
@@ -144,6 +144,7 @@ set(SRCS
detectors/rich/Unpack.cxx
detectors/rich/UnpackMS.cxx
global/ParFiles.cxx
+ global/StorableRecoResults.cxx
global/Reco.cxx
global/RecoResultsInputArchive.cxx
global/RecoResultsOutputArchive.cxx
diff --git a/algo/base/PartitionedVector.h b/algo/base/PartitionedVector.h
index 0e1c83636cb7feba358eccac7d59c3ee4c0cb336..119c2518ad15e02e20e88b727c53ebb6da1bedd9 100644
--- a/algo/base/PartitionedVector.h
+++ b/algo/base/PartitionedVector.h
@@ -142,6 +142,11 @@ namespace cbm::algo
*/
size_t NElements() const { return fData.size(); }
+ /**
+ * @brief Return total size in bytes of the underlying data.
+ */
+ size_t SizeBytes() const { return fData.size() * sizeof(T); }
+
/**
* @brief Get the underlying data.
*/
diff --git a/algo/global/Reco.cxx b/algo/global/Reco.cxx
index 3d5d594361c26e73a8ba81b52daf31ab08741e6d..d922ba9ebdfdd723f19ffabfeb71bb06d7267586 100644
--- a/algo/global/Reco.cxx
+++ b/algo/global/Reco.cxx
@@ -573,3 +573,18 @@ void Reco::QueueProcessingMetrics(const ProcessingMonitor& mon)
GetMonitor().QueueMetric("cbmreco", {{"hostname", fles::system::current_hostname()}, {"child", Opts().ChildId()}},
std::move(fields));
}
+
+void Reco::QueueProcessingExtraMetrics(const ProcessingExtraMonitor& mon)
+{
+ if (!HasMonitor()) {
+ return;
+ }
+
+ MetricFieldSet fields = {{"processingTimeIdle", FilterNan(mon.timeIdle)},
+ {"processingTimeWriteArchive", mon.timeWriteArchive.wall()},
+ {"processingThroughputWriteArchive", FilterNan(mon.timeWriteArchive.throughput())},
+ {"processingBytesWritten", FilterNan(mon.bytesWritten)}};
+
+ GetMonitor().QueueMetric("cbmreco", {{"hostname", fles::system::current_hostname()}, {"child", Opts().ChildId()}},
+ std::move(fields));
+}
diff --git a/algo/global/Reco.h b/algo/global/Reco.h
index 0496e3b3cfdb3d6804f7f82e44bf802d1a473bb1..803d770759499b2e5bc0158df265b9525ae06336 100644
--- a/algo/global/Reco.h
+++ b/algo/global/Reco.h
@@ -97,6 +97,16 @@ namespace cbm::algo
std::optional<i64> tsDelta; //< id difference between current and previous timeslice
};
+ /**
+ * @brief Monitor for additional processing steps
+ * @note Used in the main function, this should be eventually merged with ProcessingMonitor and we have a single class that handles the full processing loop
+ */
+ struct ProcessingExtraMonitor {
+ xpu::timings timeWriteArchive; //< time spent writing archive
+ size_t bytesWritten; //< bytes written to archive (estimated)
+ double timeIdle = 0.; //< time spent idle (waiting for next timeslice) [ms]
+ };
+
class Reco : SubChain {
public:
Reco();
@@ -112,6 +122,8 @@ namespace cbm::algo
void Finalize();
void PrintTimings(xpu::timings&);
+ void QueueProcessingExtraMetrics(const ProcessingExtraMonitor&);
+
private:
bool fInitialized = false;
ChainContext fContext;
diff --git a/algo/global/StorableRecoResults.cxx b/algo/global/StorableRecoResults.cxx
new file mode 100644
index 0000000000000000000000000000000000000000..5df67c5a68d11e812dafb4f01e281f25095490c8
--- /dev/null
+++ b/algo/global/StorableRecoResults.cxx
@@ -0,0 +1,35 @@
+/* Copyright (C) 2024 FIAS Frankfurt Institute for Advanced Studies, Frankfurt / Main
+ SPDX-License-Identifier: GPL-3.0-only
+ Authors: Felix Weiglhofer [committer], P.-A. Loizeau */
+#include "StorableRecoResults.h"
+
+using namespace cbm::algo;
+
+size_t StorableRecoResults::SizeBytes() const
+{
+ size_t size = 0;
+ size += fBmonDigis.size() * sizeof(CbmBmonDigi);
+ size += fStsDigis.size() * sizeof(CbmStsDigi);
+ size += fMuchDigis.size() * sizeof(CbmMuchDigi);
+ size += fTrd2dDigis.size() * sizeof(CbmTrdDigi);
+ size += fTrdDigis.size() * sizeof(CbmTrdDigi);
+ size += fTofDigis.size() * sizeof(CbmTofDigi);
+ size += fRichDigis.size() * sizeof(CbmRichDigi);
+
+ for (const auto& ev : fDigiEvents) {
+ size += ev.fData.SizeBytes();
+ }
+
+ size += fStsClusters.SizeBytes();
+ size += fStsHits.SizeBytes();
+ size += fTofHits.SizeBytes();
+ size += fTrdHits.SizeBytes();
+
+ size += fTracks.size() * sizeof(ca::Track);
+
+ // Exclude TrackHitIndexContainers for now to avoid looping over all tracks
+ // Better way to do this: Just query from boost the size of the written archive.
+ // Requires changes in flesnet to the archive classes for this
+
+ return size;
+}
diff --git a/algo/global/StorableRecoResults.h b/algo/global/StorableRecoResults.h
index 962f5b28f5e53b0fdd82131860f242757598c6ad..5560b7d1ad1998bfacc2b35e0f64335acc571820 100644
--- a/algo/global/StorableRecoResults.h
+++ b/algo/global/StorableRecoResults.h
@@ -44,6 +44,11 @@ namespace cbm::algo
*/
uint64_t TsStartTime() const { return fTsStartTime; }
+ /**
+ * @brief Total size in bytes
+ */
+ size_t SizeBytes() const;
+
std::vector<CbmBmonDigi>& BmonDigis() { return fBmonDigis; }
const std::vector<CbmBmonDigi>& BmonDigis() const { return fBmonDigis; }
diff --git a/core/data/base/CbmDigiData.h b/core/data/base/CbmDigiData.h
index 5d802e0625b34d72cb1609bee55dd4522c58577e..c864e8ba6fcef44cbdfc3a1f5f4a5886e4d2428f 100644
--- a/core/data/base/CbmDigiData.h
+++ b/core/data/base/CbmDigiData.h
@@ -31,7 +31,7 @@
**/
class CbmDigiData {
-public:
+ public:
CbmBmonDigiData fBmon; ///< Beam monitor data
CbmStsDigiData fSts; ///< STS data
CbmMuchDigiData fMuch; ///< MUCH data
@@ -94,6 +94,22 @@ public:
default: return 0; break;
}
}
+
+ /** @brief Return total size in bytes */
+ size_t SizeBytes() const
+ {
+ size_t size = 0;
+ size += fBmon.Size() * sizeof(CbmBmonDigi);
+ size += fSts.Size() * sizeof(CbmStsDigi);
+ size += fMuch.Size() * sizeof(CbmMuchDigi);
+ size += fTrd.Size() * sizeof(CbmTrdDigi);
+ size += fTrd2d.Size() * sizeof(CbmTrdDigi);
+ size += fTof.Size() * sizeof(CbmTofDigi);
+ size += fPsd.Size() * sizeof(CbmPsdDigi);
+ size += fFsd.Size() * sizeof(CbmFsdDigi);
+ size += fRich.Size() * sizeof(CbmRichDigi);
+ return size;
+ }
};
BOOST_CLASS_VERSION(CbmDigiData, 5)
diff --git a/reco/app/cbmreco/main.cxx b/reco/app/cbmreco/main.cxx
index cf1f46e520d7b571f11616c5f39d2ce8e03ee0b8..524141472e77720ce4cfa40d2e613644c2a31ab3 100644
--- a/reco/app/cbmreco/main.cxx
+++ b/reco/app/cbmreco/main.cxx
@@ -172,6 +172,8 @@ int main(int argc, char** argv)
fles::TimesliceAutoSource source(opts.InputLocator());
+ ProcessingExtraMonitor extraMonitor;
+
std::optional<RecoResultsOutputArchive> archive;
if (!opts.OutputFile().empty()) {
L_(info) << "Writing results to file: " << opts.OutputFile();
@@ -186,16 +188,26 @@ int main(int argc, char** argv)
int num_ts = opts.NumTimeslices();
if (num_ts > 0) num_ts += opts.SkipTimeslices();
L_(debug) << "Starting to fetch timeslices from source...";
+
+ auto startFetchTS = std::chrono::high_resolution_clock::now();
while (auto ts = source.get()) {
if (tsIdx < opts.SkipTimeslices()) {
tsIdx++;
continue;
}
+ auto endFetchTS = std::chrono::high_resolution_clock::now();
+ auto durationFetchTS = endFetchTS - startFetchTS;
+ extraMonitor.timeIdle +=
+ std::chrono::duration_cast<std::chrono::duration<double, std::milli>>(durationFetchTS).count();
+
try {
RecoResults result = reco.Run(*ts);
if (archive) {
+ xpu::scoped_timer t_{"Write Archive", &extraMonitor.timeWriteArchive};
auto storable = makeStorableRecoResults(*ts, result);
+ extraMonitor.bytesWritten = storable->SizeBytes();
+ xpu::t_add_bytes(extraMonitor.bytesWritten);
archive->put(storable);
}
}
@@ -203,6 +215,7 @@ int main(int argc, char** argv)
// TODO: Add flag if we want to abort on exception or continue with next timeslice
L_(error) << "Caught ProcessingError while processing timeslice " << tsIdx << ": " << e.what();
}
+ reco.QueueProcessingExtraMetrics(extraMonitor);
// Release memory after each timeslice and log memory usage
// This is useful to detect memory leaks as the memory usage should be constant between timeslices
@@ -212,6 +225,8 @@ int main(int argc, char** argv)
tsIdx++;
if (num_ts > 0 && tsIdx >= num_ts) break;
+
+ startFetchTS = std::chrono::high_resolution_clock::now();
}
if (archive) archive->end_stream();