Skip to content

Commit ce7127e

Browse files
sawenzelclaude
sawenzel
and
claude
committed
Publish an empty timeframe where the readers only ended the stream
This makes the reader specs follow one policy for a tree without an entry: publish empty output and then end the stream. - 36 readers ended the stream without sending anything, while the ITS/MFT, MCH and MID digit readers sent empty output first. - A reader that ends the stream without publishing leaves its own writer with a zero-entry tree, which recreates the same condition in the next step. - Readers whose branch data lives in owned members now skip only the GetEntry; the ones binding ROOT-allocated pointers publish an explicit empty container, since those pointers are null before the first GetEntry. - The TPC track reader called accumulate(), which reads the entry, before the check. https://its.cern.ch/jira/browse/O2-7132 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1 parent a787ad2 commit ce7127e

36 files changed

Lines changed: 491 additions & 440 deletions

‎Detectors/CPV/workflow/src/ClusterReaderSpec.cxx‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -41,25 +41,25 @@ void ClusterReader::init(InitContext& ic)
4141
void ClusterReader::run(ProcessingContext& pc)
4242
{
4343
auto ent = mTree->GetReadEntry() + 1;
44-
if (ent >= mTree->GetEntries()) {
45-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
46-
// the tree then has no entry to read. End the stream instead of reading past the end and
47-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
48-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
49-
LOG(info) << "no entry to read, ending the stream";
50-
pc.services().get<ControlService>().endOfStream();
51-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
52-
return;
44+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
45+
// the tree then has no entry to read. Publish empty containers instead of reading past the
46+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
47+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
48+
// every production build since ENABLE_CASSERT defaults to OFF.)
49+
const bool noEntry = ent >= mTree->GetEntries();
50+
if (noEntry) {
51+
LOG(info) << "no entry to read, publishing empty output";
52+
} else {
53+
mTree->GetEntry(ent);
5354
}
54-
mTree->GetEntry(ent);
5555
LOG(info) << "Pushing " << mClusters.size() << " Clusters in " << mTRs.size() << " TriggerRecords at entry " << ent;
5656
pc.outputs().snapshot(Output{mOrigin, "CLUSTERS", 0}, mClusters);
5757
pc.outputs().snapshot(Output{mOrigin, "CLUSTERTRIGRECS", 0}, mTRs);
5858
if (mUseMC) {
5959
pc.outputs().snapshot(Output{mOrigin, "CLUSTERTRUEMC", 0}, mMCTruth);
6060
}
6161

62-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
62+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
6363
pc.services().get<ControlService>().endOfStream();
6464
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
6565
}

‎Detectors/CPV/workflow/src/DigitReaderSpec.cxx‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -41,25 +41,25 @@ void DigitReader::init(InitContext& ic)
4141
void DigitReader::run(ProcessingContext& pc)
4242
{
4343
auto ent = mTree->GetReadEntry() + 1;
44-
if (ent >= mTree->GetEntries()) {
45-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
46-
// the tree then has no entry to read. End the stream instead of reading past the end and
47-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
48-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
49-
LOG(info) << "no entry to read, ending the stream";
50-
pc.services().get<ControlService>().endOfStream();
51-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
52-
return;
44+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
45+
// the tree then has no entry to read. Publish empty containers instead of reading past the
46+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
47+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
48+
// every production build since ENABLE_CASSERT defaults to OFF.)
49+
const bool noEntry = ent >= mTree->GetEntries();
50+
if (noEntry) {
51+
LOG(info) << "no entry to read, publishing empty output";
52+
} else {
53+
mTree->GetEntry(ent);
5354
}
54-
mTree->GetEntry(ent);
5555
LOG(info) << "Pushing " << mDigits.size() << " Digits in " << mTRs.size() << " TriggerRecords at entry " << ent;
5656
pc.outputs().snapshot(Output{mOrigin, "DIGITS", 0}, mDigits);
5757
pc.outputs().snapshot(Output{mOrigin, "DIGITTRIGREC", 0}, mTRs);
5858
if (mUseMC) {
5959
pc.outputs().snapshot(Output{mOrigin, "DIGITSMCTR", 0}, mMCTruth);
6060
}
6161

62-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
62+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
6363
pc.services().get<ControlService>().endOfStream();
6464
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
6565
}

‎Detectors/CTP/workflowIO/src/DigitReaderSpec.cxx‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -86,21 +86,21 @@ void DigitReader::run(ProcessingContext& pc)
8686
auto ent = mTree->GetReadEntry();
8787
if (!mUseIRFrames) {
8888
ent++;
89-
if (ent >= mTree->GetEntries()) {
90-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
91-
// the tree then has no entry to read. End the stream instead of reading past the end and
92-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
93-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
94-
LOG(info) << "no entry to read, ending the stream";
95-
pc.services().get<ControlService>().endOfStream();
96-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
97-
return;
89+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
90+
// the tree then has no entry to read. Publish empty containers instead of reading past the
91+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
92+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
93+
// every production build since ENABLE_CASSERT defaults to OFF.)
94+
const bool noEntry = ent >= mTree->GetEntries();
95+
if (noEntry) {
96+
LOG(info) << "no entry to read, publishing empty output";
97+
} else {
98+
mTree->GetEntry(ent);
9899
}
99-
mTree->GetEntry(ent);
100100
LOG(info) << "DigitReader pushes " << mDigits.size() << " digits at entry " << ent;
101101
pc.outputs().snapshot(Output{"CTP", "DIGITS", 0}, mDigits);
102102
pc.outputs().snapshot(Output{"CTP", "LUMI", 0}, mLumi);
103-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
103+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
104104
pc.services().get<ControlService>().endOfStream();
105105
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
106106
}

‎Detectors/FIT/FDD/workflow/src/DigitReaderSpec.cxx‎

Lines changed: 25 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -77,33 +77,40 @@ void DigitReader::run(ProcessingContext& pc)
7777
}
7878
}
7979
auto ent = mTree->GetReadEntry() + 1;
80-
if (ent >= mTree->GetEntries()) {
81-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
82-
// the tree then has no entry to read. End the stream instead of reading past the end and
83-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
84-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
85-
LOG(info) << "no entry to read, ending the stream";
86-
pc.services().get<ControlService>().endOfStream();
87-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
88-
return;
80+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
81+
// the tree then has no entry to read. Publish empty containers instead of reading past the
82+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
83+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
84+
// every production build since ENABLE_CASSERT defaults to OFF.)
85+
const bool noEntry = ent >= mTree->GetEntries();
86+
if (noEntry) {
87+
LOG(info) << "no entry to read, publishing empty output";
88+
} else {
89+
mTree->GetEntry(ent);
8990
}
90-
mTree->GetEntry(ent);
9191

92-
LOG(info) << "FDD DigitReader pushes " << digitsBC->size() << " digits";
93-
pc.outputs().snapshot(Output{mOrigin, "DIGITSBC", 0}, *digitsBC);
94-
pc.outputs().snapshot(Output{mOrigin, "DIGITSCH", 0}, *digitsCh);
92+
static const std::vector<o2::fdd::Digit> noDigitsBC;
93+
static const std::vector<o2::fdd::ChannelData> noDigitsCh;
94+
static const std::vector<o2::fdd::DetTrigInput> noDigitsTrig;
95+
const auto& digitsBCOut = noEntry ? noDigitsBC : *digitsBC;
96+
const auto& digitsChOut = noEntry ? noDigitsCh : *digitsCh;
97+
LOG(info) << "FDD DigitReader pushes " << digitsBCOut.size() << " digits";
98+
pc.outputs().snapshot(Output{mOrigin, "DIGITSBC", 0}, digitsBCOut);
99+
pc.outputs().snapshot(Output{mOrigin, "DIGITSCH", 0}, digitsChOut);
95100

96101
if (mUseMC) {
97102
// TODO: To be replaced with sending ConstMCTruthContainer as soon as reco workflow supports it
98-
pc.outputs().snapshot(Output{mOrigin, "TRIGGERINPUT", 0}, *digitsTrig);
103+
pc.outputs().snapshot(Output{mOrigin, "TRIGGERINPUT", 0}, noEntry ? noDigitsTrig : *digitsTrig);
99104

100-
std::vector<char> flatbuffer;
101-
mcTruthRootBuffer->copyandflatten(flatbuffer);
102105
o2::dataformats::MCTruthContainer<o2::fdd::MCLabel> mcTruth;
103-
mcTruth.restore_from(flatbuffer.data(), flatbuffer.size());
106+
if (!noEntry) {
107+
std::vector<char> flatbuffer;
108+
mcTruthRootBuffer->copyandflatten(flatbuffer);
109+
mcTruth.restore_from(flatbuffer.data(), flatbuffer.size());
110+
}
104111
pc.outputs().snapshot(Output{mOrigin, "DIGITLBL", 0}, mcTruth);
105112
}
106-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
113+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
107114
pc.services().get<ControlService>().endOfStream();
108115
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
109116
}

‎Detectors/FIT/FDD/workflow/src/RecPointReaderSpec.cxx‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -45,23 +45,27 @@ void RecPointReader::init(InitContext& ic)
4545
void RecPointReader::run(ProcessingContext& pc)
4646
{
4747
auto ent = mTree->GetReadEntry() + 1;
48-
if (ent >= mTree->GetEntries()) {
49-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
50-
// the tree then has no entry to read. End the stream instead of reading past the end and
51-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
52-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
53-
LOG(info) << "no entry to read, ending the stream";
54-
pc.services().get<ControlService>().endOfStream();
55-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
56-
return;
48+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
49+
// the tree then has no entry to read. Publish empty containers instead of reading past the
50+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
51+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
52+
// every production build since ENABLE_CASSERT defaults to OFF.)
53+
const bool noEntry = ent >= mTree->GetEntries();
54+
if (noEntry) {
55+
LOG(info) << "no entry to read, publishing empty output";
56+
} else {
57+
mTree->GetEntry(ent);
5758
}
58-
mTree->GetEntry(ent);
5959

60-
LOG(info) << "FDD RecPointReader pushes " << mRecPoints->size() << " recpoints with " << mChannelData->size() << " channels at entry " << ent;
61-
pc.outputs().snapshot(Output{mOrigin, "RECPOINTS", 0}, *mRecPoints);
62-
pc.outputs().snapshot(Output{mOrigin, "RECCHDATA", 0}, *mChannelData);
60+
static const std::vector<o2::fdd::RecPoint> noRecPoints;
61+
static const std::vector<o2::fdd::ChannelDataFloat> noChannelData;
62+
const auto& recPoints = noEntry ? noRecPoints : *mRecPoints;
63+
const auto& channelData = noEntry ? noChannelData : *mChannelData;
64+
LOG(info) << "FDD RecPointReader pushes " << recPoints.size() << " recpoints with " << channelData.size() << " channels at entry " << ent;
65+
pc.outputs().snapshot(Output{mOrigin, "RECPOINTS", 0}, recPoints);
66+
pc.outputs().snapshot(Output{mOrigin, "RECCHDATA", 0}, channelData);
6367

64-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
68+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
6569
pc.services().get<ControlService>().endOfStream();
6670
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
6771
}

‎Detectors/FIT/FT0/workflow/src/DigitReaderSpec.cxx‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -61,17 +61,17 @@ void DigitReader::run(ProcessingContext& pc)
6161
mTree->SetBranchAddress("FT0DIGITSMCTR", &plabels);
6262
}
6363
auto ent = mTree->GetReadEntry() + 1;
64-
if (ent >= mTree->GetEntries()) {
65-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
66-
// the tree then has no entry to read. End the stream instead of reading past the end and
67-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
68-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
69-
LOG(info) << "no entry to read, ending the stream";
70-
pc.services().get<ControlService>().endOfStream();
71-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
72-
return;
64+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
65+
// the tree then has no entry to read. Publish empty containers instead of reading past the
66+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
67+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
68+
// every production build since ENABLE_CASSERT defaults to OFF.)
69+
const bool noEntry = ent >= mTree->GetEntries();
70+
if (noEntry) {
71+
LOG(info) << "no entry to read, publishing empty output";
72+
} else {
73+
mTree->GetEntry(ent);
7374
}
74-
mTree->GetEntry(ent);
7575
LOG(debug) << "FT0DigitReader pushed " << channels.size() << " channels in " << digits.size() << " digits";
7676
pc.outputs().snapshot(Output{"FT0", "DIGITSBC", 0}, digits);
7777
pc.outputs().snapshot(Output{"FT0", "DIGITSCH", 0}, channels);
@@ -81,7 +81,7 @@ void DigitReader::run(ProcessingContext& pc)
8181
if (mUseTrgInput) {
8282
pc.outputs().snapshot(Output{"FT0", "TRIGGERINPUT", 0}, trgInput);
8383
}
84-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
84+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
8585
pc.services().get<ControlService>().endOfStream();
8686
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
8787
}

‎Detectors/FIT/FT0/workflow/src/RecPointReaderSpec.cxx‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -45,23 +45,27 @@ void RecPointReader::init(InitContext& ic)
4545
void RecPointReader::run(ProcessingContext& pc)
4646
{
4747
auto ent = mTree->GetReadEntry() + 1;
48-
if (ent >= mTree->GetEntries()) {
49-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
50-
// the tree then has no entry to read. End the stream instead of reading past the end and
51-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
52-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
53-
LOG(info) << "no entry to read, ending the stream";
54-
pc.services().get<ControlService>().endOfStream();
55-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
56-
return;
48+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
49+
// the tree then has no entry to read. Publish empty containers instead of reading past the
50+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
51+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
52+
// every production build since ENABLE_CASSERT defaults to OFF.)
53+
const bool noEntry = ent >= mTree->GetEntries();
54+
if (noEntry) {
55+
LOG(info) << "no entry to read, publishing empty output";
56+
} else {
57+
mTree->GetEntry(ent);
5758
}
58-
mTree->GetEntry(ent);
5959

60-
LOG(debug) << "FT0 RecPointReader pushes " << mRecPoints->size() << " recpoints with " << mChannelData->size() << " channels at entry " << ent;
61-
pc.outputs().snapshot(Output{mOrigin, "RECPOINTS", 0}, *mRecPoints);
62-
pc.outputs().snapshot(Output{mOrigin, "RECCHDATA", 0}, *mChannelData);
60+
static const std::vector<o2::ft0::RecPoints> noRecPoints;
61+
static const std::vector<o2::ft0::ChannelDataFloat> noChannelData;
62+
const auto& recPoints = noEntry ? noRecPoints : *mRecPoints;
63+
const auto& channelData = noEntry ? noChannelData : *mChannelData;
64+
LOG(debug) << "FT0 RecPointReader pushes " << recPoints.size() << " recpoints with " << channelData.size() << " channels at entry " << ent;
65+
pc.outputs().snapshot(Output{mOrigin, "RECPOINTS", 0}, recPoints);
66+
pc.outputs().snapshot(Output{mOrigin, "RECCHDATA", 0}, channelData);
6367

64-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
68+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
6569
pc.services().get<ControlService>().endOfStream();
6670
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
6771
}

‎Detectors/FIT/FV0/workflow/src/DigitReaderSpec.cxx‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -62,17 +62,17 @@ void DigitReader::run(ProcessingContext& pc)
6262
mTree->SetBranchAddress("FV0DigitLabels", &plabels);
6363
}
6464
auto ent = mTree->GetReadEntry() + 1;
65-
if (ent >= mTree->GetEntries()) {
66-
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
67-
// the tree then has no entry to read. End the stream instead of reading past the end and
68-
// publishing branch addresses that GetEntry has not filled. This was an assert, which is
69-
// compiled out of every production build since ENABLE_CASSERT defaults to OFF.
70-
LOG(info) << "no entry to read, ending the stream";
71-
pc.services().get<ControlService>().endOfStream();
72-
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
73-
return;
65+
// A timeframe holds no collision at all whenever the interaction rate is low enough, and
66+
// the tree then has no entry to read. Publish empty containers instead of reading past the
67+
// end and pushing branch addresses that GetEntry has not filled, so that the consumers
68+
// downstream still see the timeframe. (This used to be an assert, which is compiled out of
69+
// every production build since ENABLE_CASSERT defaults to OFF.)
70+
const bool noEntry = ent >= mTree->GetEntries();
71+
if (noEntry) {
72+
LOG(info) << "no entry to read, publishing empty output";
73+
} else {
74+
mTree->GetEntry(ent);
7475
}
75-
mTree->GetEntry(ent);
7676
LOG(debug) << "FV0DigitReader pushed " << channels.size() << " channels in " << digits.size() << " digits";
7777
pc.outputs().snapshot(Output{"FV0", "DIGITSBC", 0}, digits);
7878
pc.outputs().snapshot(Output{"FV0", "DIGITSCH", 0}, channels);
@@ -82,7 +82,7 @@ void DigitReader::run(ProcessingContext& pc)
8282
if (mUseTrgInput) {
8383
pc.outputs().snapshot(Output{"FV0", "TRIGGERINPUT", 0}, trgInput);
8484
}
85-
if (mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
85+
if (noEntry || mTree->GetReadEntry() + 1 >= mTree->GetEntries()) {
8686
pc.services().get<ControlService>().endOfStream();
8787
pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);
8888
}

0 commit comments

Comments
 (0)