Skip to content

Commit e2962cb

Browse files
matthiasrichtersawenzel
matthiasrichter
authored and
sawenzel
committed
Make the processing function of TPC Clusterer and Decoder processes more flexible
Processing function is now agnostic of the definition of inputs. Processing is done on the bases of a loop over all available inputs rather than checking specific input bindings.
1 parent befe811 commit e2962cb

2 files changed

Lines changed: 101 additions & 51 deletions

File tree

‎Detectors/TPC/workflow/src/ClusterDecoderRawSpec.cxx‎

Lines changed: 50 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -49,25 +49,19 @@ using MCLabelContainer = o2::dataformats::MCTruthContainer<o2::MCCompLabel>;
4949
/// MC labels are received as MCLabelContainers
5050
DataProcessorSpec getClusterDecoderRawSpec(bool sendMC)
5151
{
52+
using DataDescription = o2::header::DataDescription;
5253
std::string processorName = "tpc-cluster-decoder";
5354

54-
auto initFunction = [sendMC](InitContext& ic) {
55+
auto initFunction = [](InitContext& ic) {
5556
// there is nothing to init at the moment
5657
auto verbosity = 0;
5758
auto decoder = std::make_shared<HardwareClusterDecoder>();
5859

59-
auto processingFct = [verbosity, decoder, sendMC](ProcessingContext& pc) {
60-
static bool finished = false;
61-
if (finished) {
62-
return;
63-
}
60+
auto processSectorFunction = [verbosity, decoder](ProcessingContext& pc, std::string inputKey, std::string labelKey) -> bool {
6461
// this will return a span of TPC clusters
65-
const auto& ref = pc.inputs().get("rawin");
62+
const auto& ref = pc.inputs().get(inputKey.c_str());
6663
auto size = o2::framework::DataRefUtils::getPayloadSize(ref);
67-
if (ref.payload == nullptr) {
68-
return;
69-
}
70-
auto const* dataHeader = DataRefUtils::getHeader<o2::header::DataHeader*>(pc.inputs().get("rawin"));
64+
auto const* dataHeader = DataRefUtils::getHeader<o2::header::DataHeader*>(pc.inputs().get(inputKey.c_str()));
7165
o2::header::DataHeader::SubSpecificationType fanSpec = dataHeader->subSpecification;
7266

7367
// init the stacks for forwarding the sector header
@@ -76,28 +70,23 @@ DataProcessorSpec getClusterDecoderRawSpec(bool sendMC)
7670
o2::header::Stack rawHeaderStack;
7771
o2::header::Stack mcHeaderStack;
7872
o2::TPC::TPCSectorHeader const* sectorHeaderMC = nullptr;
79-
if (sendMC) {
80-
sectorHeaderMC = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(pc.inputs().get("mclblin"));
73+
if (!labelKey.empty()) {
74+
sectorHeaderMC = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(pc.inputs().get(labelKey.c_str()));
8175
if (sectorHeaderMC) {
8276
o2::header::Stack actual{ *sectorHeaderMC };
8377
std::swap(mcHeaderStack, actual);
8478
if (sectorHeaderMC->sector < 0) {
85-
pc.outputs().snapshot(OutputRef{ "mclblout", fanSpec, std::move(mcHeaderStack) }, fanSpec);
79+
pc.outputs().snapshot(Output{ gDataOriginTPC, DataDescription("CLNATIVEMCLBL"), fanSpec, Lifetime::Timeframe, std::move(mcHeaderStack) }, fanSpec);
8680
}
8781
}
8882
}
89-
auto const* sectorHeader = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(pc.inputs().get("rawin"));
83+
auto const* sectorHeader = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(pc.inputs().get(inputKey.c_str()));
9084
if (sectorHeader) {
9185
o2::header::Stack actual{ *sectorHeader };
9286
std::swap(rawHeaderStack, actual);
9387
if (sectorHeader->sector < 0) {
94-
pc.outputs().snapshot(OutputRef{ "clusterout", fanSpec, std::move(rawHeaderStack) }, fanSpec);
95-
if (sectorHeader->sector == -1) {
96-
// got EOD
97-
finished = true;
98-
pc.services().get<ControlService>().readyToQuit(false);
99-
}
100-
return;
88+
pc.outputs().snapshot(Output{ gDataOriginTPC, DataDescription("CLUSTERNATIVE"), fanSpec, Lifetime::Timeframe, std::move(rawHeaderStack) }, fanSpec);
89+
return (sectorHeader->sector == -1);
10190
}
10291
}
10392
assert(sectorHeaderMC == nullptr || sectorHeader->sector == sectorHeaderMC->sector);
@@ -113,8 +102,8 @@ DataProcessorSpec getClusterDecoderRawSpec(bool sendMC)
113102
// in the raw pages
114103
std::vector<MCLabelContainer> mcinCopies;
115104
std::unique_ptr<const MCLabelContainer> mcin;
116-
if (sendMC) {
117-
mcin = std::move(pc.inputs().get<MCLabelContainer*>("mclblin"));
105+
if (!labelKey.empty()) {
106+
mcin = std::move(pc.inputs().get<MCLabelContainer*>(labelKey.c_str()));
118107
mcinCopies.resize(nPages);
119108
if (verbosity > 0) {
120109
LOG(INFO) << "Decoder input: " << size << ", " << nPages << " pages, " << mcin->getIndexedSize() << " MC label sets";
@@ -171,7 +160,7 @@ DataProcessorSpec getClusterDecoderRawSpec(bool sendMC)
171160
totalSize += coll.getFlatSize() + sizeof(ClusterGroupHeader) - sizeof(ClusterGroupAttribute);
172161
}
173162

174-
auto* target = pc.outputs().newChunk(OutputRef{ "clusterout", fanSpec, std::move(rawHeaderStack) }, totalSize).data;
163+
auto* target = pc.outputs().newChunk(Output{ gDataOriginTPC, DataDescription("CLUSTERNATIVE"), fanSpec, Lifetime::Timeframe, std::move(rawHeaderStack) }, totalSize).data;
175164

176165
for (const auto& coll : cont) {
177166
if (verbosity > 1) {
@@ -185,14 +174,48 @@ DataProcessorSpec getClusterDecoderRawSpec(bool sendMC)
185174
memcpy(target, coll.data(), coll.getFlatSize() - sizeof(ClusterGroupAttribute));
186175
target += coll.getFlatSize() - sizeof(ClusterGroupAttribute);
187176
}
188-
if (sendMC) {
177+
if (!labelKey.empty()) {
189178
if (verbosity > 0) {
190179
LOG(INFO) << "sending " << mcoutList.size() << " MC label container(s) with in total "
191180
<< std::accumulate(mcoutList.begin(), mcoutList.end(), size_t(0), [](size_t l, auto const& r) { return l + r.getIndexedSize(); })
192181
<< " label object(s)" << std::endl;
193182
}
194183
// serialize the complete list of MC label containers
195-
pc.outputs().snapshot(OutputRef{ "mclblout", fanSpec, std::move(mcHeaderStack) }, mcoutList);
184+
pc.outputs().snapshot(Output{ gDataOriginTPC, DataDescription("CLNATIVEMCLBL"), fanSpec, Lifetime::Timeframe, std::move(mcHeaderStack) }, mcoutList);
185+
}
186+
return false;
187+
};
188+
189+
auto processingFct = [verbosity, decoder, processSectorFunction](ProcessingContext& pc) {
190+
static bool finished = false;
191+
if (finished) {
192+
return;
193+
}
194+
195+
struct SectorInputDesc {
196+
std::string inputKey = "";
197+
std::string labelKey = "";
198+
};
199+
std::map<o2::header::DataHeader::SubSpecificationType, SectorInputDesc> inputs;
200+
for (auto const& inputRef : pc.inputs()) {
201+
auto const* dataHeader = DataRefUtils::getHeader<o2::header::DataHeader*>(inputRef);
202+
assert(dataHeader);
203+
if (dataHeader->dataOrigin == gDataOriginTPC && dataHeader->dataDescription == DataDescription("CLUSTERHW")) {
204+
inputs[dataHeader->subSpecification].inputKey = inputRef.spec->binding;
205+
} else if (dataHeader->dataOrigin == gDataOriginTPC && dataHeader->dataDescription == DataDescription("CLUSTERHWMCLBL")) {
206+
inputs[dataHeader->subSpecification].labelKey = inputRef.spec->binding;
207+
}
208+
}
209+
// will stay true if all inputs signal finished
210+
// this implies that all inputs are always processed together, when changing this policy the
211+
// status of each input needs to be kept individually
212+
finished = true;
213+
for (auto const& input : inputs) {
214+
finished = finished & processSectorFunction(pc, input.second.inputKey, input.second.labelKey);
215+
}
216+
if (finished) {
217+
// got EOD on all inputs
218+
pc.services().get<ControlService>().readyToQuit(false);
196219
}
197220
};
198221

‎Detectors/TPC/workflow/src/ClustererSpec.cxx‎

Lines changed: 51 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
#include <FairMQLogger.h>
2727
#include <memory> // for make_shared
2828
#include <vector>
29+
#include <numeric> // std::accumulate
2930

3031
using namespace o2::framework;
3132
using namespace o2::header;
@@ -52,26 +53,23 @@ DataProcessorSpec getClustererSpec(bool sendMC)
5253
bool finished = false;
5354
};
5455

55-
auto initFunction = [sendMC](InitContext& ic) {
56+
auto initFunction = [](InitContext& ic) {
5657
// FIXME: the clusterer needs to be initialized with the sector number, so we need one
5758
// per sector. Taking a closer look to the HwClusterer, the sector number is only used
5859
// for calculating the CRU id. This could be achieved by passing the current sector as
5960
// parameter to the clusterer processing function.
6061
auto processAttributes = std::make_shared<ProcessAttributes>();
6162

62-
auto processingFct = [processAttributes, sendMC](ProcessingContext& pc) {
63-
if (processAttributes->finished) {
64-
return;
65-
}
63+
auto processSectorFunction = [processAttributes](ProcessingContext& pc, std::string inputKey, std::string labelKey) -> bool {
6664
auto& clusterArray = processAttributes->clusterArray;
6765
auto& mctruthArray = processAttributes->mctruthArray;
6866
auto& clusterers = processAttributes->clusterers;
6967
auto& verbosity = processAttributes->verbosity;
70-
auto dataref = pc.inputs().get("digits");
68+
auto dataref = pc.inputs().get(inputKey);
7169
auto const* sectorHeader = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(dataref);
7270
if (sectorHeader == nullptr) {
7371
LOG(ERROR) << "sector header missing on header stack";
74-
return;
72+
return false;
7573
}
7674
auto const* dataHeader = DataRefUtils::getHeader<o2::header::DataHeader*>(dataref);
7775
o2::header::DataHeader::SubSpecificationType fanSpec = dataHeader->subSpecification;
@@ -81,22 +79,17 @@ DataProcessorSpec getClustererSpec(bool sendMC)
8179
// forward the control information
8280
// FIXME define and use flags in TPCSectorHeader
8381
o2::TPC::TPCSectorHeader header{ sector };
84-
pc.outputs().snapshot(OutputRef{ "clusters", fanSpec, { header } }, fanSpec);
85-
if (sendMC) {
86-
pc.outputs().snapshot(OutputRef{ "clusterlbl", fanSpec, { header } }, fanSpec);
82+
pc.outputs().snapshot(Output{ gDataOriginTPC, "CLUSTERSIM", fanSpec, Lifetime::Timeframe, { header } }, fanSpec);
83+
if (!labelKey.empty()) {
84+
pc.outputs().snapshot(Output{ gDataOriginTPC, "CLUSTERMCLBL", fanSpec, Lifetime::Timeframe, { header } }, fanSpec);
8785
}
88-
if (sectorHeader->sector == -1) {
89-
// got EOD
90-
processAttributes->finished = true;
91-
pc.services().get<ControlService>().readyToQuit(false);
92-
}
93-
return;
86+
return (sectorHeader->sector == -1);
9487
}
9588
std::unique_ptr<const MCLabelContainer> inMCLabels;
96-
if (sendMC) {
97-
inMCLabels = std::move(pc.inputs().get<const MCLabelContainer*>("mclabels"));
89+
if (!labelKey.empty()) {
90+
inMCLabels = std::move(pc.inputs().get<const MCLabelContainer*>(labelKey.c_str()));
9891
}
99-
auto inDigits = pc.inputs().get<const std::vector<o2::TPC::Digit>>("digits");
92+
auto inDigits = pc.inputs().get<const std::vector<o2::TPC::Digit>>(inputKey.c_str());
10093
if (verbosity > 0 && inMCLabels) {
10194
LOG(INFO) << "received " << inDigits.size() << " digits, "
10295
<< inMCLabels->getIndexedSize() << " MC label objects";
@@ -121,17 +114,51 @@ DataProcessorSpec getClustererSpec(bool sendMC)
121114
const std::vector<o2::TPC::Digit> emptyDigits;
122115
clusterer->finishProcess(emptyDigits, nullptr, false); // keep here the false, otherwise the clusters are lost of they are not stored in the meantime
123116
if (verbosity > 0) {
124-
LOG(INFO) << "clusterer produced " << clusterArray.size() << " cluster(s)";
125-
if (sendMC) {
117+
LOG(INFO) << "clusterer produced "
118+
<< std::accumulate(clusterArray.begin(), clusterArray.end(), size_t(0), [](size_t l, auto const& r) { return l + r.getContainer()->numberOfClusters; })
119+
<< " cluster(s)";
120+
if (!labelKey.empty()) {
126121
LOG(INFO) << "clusterer produced " << mctruthArray.getIndexedSize() << " MC label object(s)";
127122
}
128123
}
129-
pc.outputs().snapshot(OutputRef{ "clusters", fanSpec, { *sectorHeader } }, clusterArray);
130-
if (sendMC) {
131-
pc.outputs().snapshot(OutputRef{ "clusterlbl", fanSpec, { *sectorHeader } }, mctruthArray);
124+
pc.outputs().snapshot(Output{ gDataOriginTPC, "CLUSTERSIM", fanSpec, Lifetime::Timeframe, { *sectorHeader } }, clusterArray);
125+
if (!labelKey.empty()) {
126+
pc.outputs().snapshot(Output{ gDataOriginTPC, "CLUSTERMCLBL", fanSpec, Lifetime::Timeframe, { *sectorHeader } }, mctruthArray);
132127
}
128+
return false;
133129
};
134130

131+
auto processingFct = [processAttributes, processSectorFunction](ProcessingContext& pc) {
132+
if (processAttributes->finished) {
133+
return;
134+
}
135+
136+
struct SectorInputDesc {
137+
std::string inputKey = "";
138+
std::string labelKey = "";
139+
};
140+
std::map<o2::header::DataHeader::SubSpecificationType, SectorInputDesc> inputs;
141+
for (auto const& inputRef : pc.inputs()) {
142+
auto const* dataHeader = DataRefUtils::getHeader<o2::header::DataHeader*>(inputRef);
143+
assert(dataHeader);
144+
if (dataHeader->dataOrigin == gDataOriginTPC && dataHeader->dataDescription == o2::header::DataDescription("DIGITS")) {
145+
inputs[dataHeader->subSpecification].inputKey = inputRef.spec->binding;
146+
} else if (dataHeader->dataOrigin == gDataOriginTPC && dataHeader->dataDescription == o2::header::DataDescription("DIGITSMCTR")) {
147+
inputs[dataHeader->subSpecification].labelKey = inputRef.spec->binding;
148+
}
149+
}
150+
bool finished = true;
151+
for (auto const& input : inputs) {
152+
if (!processSectorFunction(pc, input.second.inputKey, input.second.labelKey)) {
153+
finished = false;
154+
}
155+
}
156+
if (finished) {
157+
// got EOD on all inputs
158+
processAttributes->finished = true;
159+
pc.services().get<ControlService>().readyToQuit(false);
160+
}
161+
};
135162
return processingFct;
136163
};
137164

0 commit comments

Comments
 (0)