Skip to content

Commit b3ab7f4

Browse files
matthiasrichtersawenzel
matthiasrichter
authored and
sawenzel
committed
Implementing parallel processing in the tpc-reco-workflow
Moving TPC reco workflow definition to separate file and creating library for module TPCWorkflow with reco workflow and processor specs. Parallel workflow can be configured using option '--tpc-lanes n' Added features: - workflow option --tpc-lanes - DigitReaderSpec fan out - CATrackerSpec fan in - adding input type 'digitizer' to connect directly to the digitizer workflow - adjusting data descriptions to output of digitizer workflow ('DIGITS', 'DIGITSMCTR'). - forwarding empty data sets at end of processing. - adjusting processor names: prefix tpc
1 parent aae012f commit b3ab7f4

15 files changed

Lines changed: 446 additions & 238 deletions

‎Detectors/TPC/workflow/CMakeLists.txt‎

Lines changed: 13 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -12,28 +12,30 @@ set(MODULE_NAME "TPCWorkflow")
1212
set(MODULE_BUCKET_NAME TPC_workflow_bucket)
1313

1414
O2_SETUP(NAME ${MODULE_NAME})
15-
#set(SRCS
16-
# )
15+
set(SRCS
16+
src/RecoWorkflow.cxx
17+
src/DigitReaderSpec.cxx
18+
src/ClusterReaderSpec.cxx
19+
src/ClustererSpec.cxx
20+
src/ClusterConverterSpec.cxx
21+
src/ClusterDecoderRawSpec.cxx
22+
src/CATrackerSpec.cxx
23+
src/RootFileWriterSpec.cxx
24+
)
1725

1826
## TODO: feature of macro, it deletes the variables we pass to it, set them again
1927
## this has to be fixed in the macro implementation
2028
set(LIBRARY_NAME ${MODULE_NAME})
2129
set(BUCKET_NAME ${MODULE_BUCKET_NAME})
2230

23-
#O2_GENERATE_LIBRARY()
31+
O2_GENERATE_LIBRARY()
2432

2533
O2_GENERATE_EXECUTABLE(
2634
EXE_NAME tpc-reco-workflow
2735

2836
SOURCES
2937
src/tpc-reco-workflow.cxx
30-
src/DigitReaderSpec.cxx
31-
src/ClusterReaderSpec.cxx
32-
src/ClustererSpec.cxx
33-
src/ClusterConverterSpec.cxx
34-
src/ClusterDecoderRawSpec.cxx
35-
src/CATrackerSpec.cxx
36-
src/RootFileWriterSpec.cxx
3738

38-
BUCKET_NAME ${MODULE_BUCKET_NAME}
39+
MODULE_LIBRARY_NAME ${LIBRARY_NAME}
40+
BUCKET_NAME ${BUCKET_NAME}
3941
)

‎Detectors/TPC/workflow/README.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,10 +53,15 @@ tpc-reco-workflow --infile tpcdigits.root --tpc-sectors 0-15 --disable-mc 1
5353
--input-type arg (=digits) digits, clusters, raw
5454
--output-type arg (=tracks) clusters, raw, tracks
5555
--disable-mc arg (=0) disable sending of MC information
56+
--tpc-lanes arg (=1) number of parallel lanes up to the tracker
5657
```
5758
Support for all other output types than `tracks` is going to be implemented soon, multiple outputs
5859
will be supported in order to keep the data at intermediate steps.
5960

61+
#### Parallel processing
62+
Parallel processing is controlled by the option `--tpc-lanes n`. The digit reader will fan out to n processing
63+
lanes, each with clusterer, converter and decoder. The tracker will fan in from the parallel lanes.
64+
6065
### Processor options
6166

6267
#### TPC CA tracker

‎Detectors/TPC/workflow/include/TPCWorkflow/RecoWorkflow.h‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
// Copyright CERN and copyright holders of ALICE O2. This software is
2+
// distributed under the terms of the GNU General Public License v3 (GPL
3+
// Version 3), copied verbatim in the file "COPYING".
4+
//
5+
// See http://alice-o2.web.cern.ch/license for full licensing information.
6+
//
7+
// In applying this license CERN does not waive the privileges and immunities
8+
// granted to it by virtue of its status as an Intergovernmental Organization
9+
// or submit itself to any jurisdiction.
10+
11+
#ifndef O2_TPC_RECOWORKFLOW_H
12+
#define O2_TPC_RECOWORKFLOW_H
13+
/// @file RecoWorkflow.h
14+
/// @author Matthias Richter
15+
/// @since 2018-09-26
16+
/// @brief Workflow definition for the TPC reconstruction
17+
18+
#include "Framework/WorkflowSpec.h"
19+
#include <string>
20+
namespace o2
21+
{
22+
namespace TPC
23+
{
24+
25+
namespace RecoWorkflow
26+
{
27+
/// define input and output types of the workflow
28+
enum struct InputType { Digitizer, // directly read digits from {TPC:DIGITS}
29+
Digits, // read digits from file
30+
Clusters, // read clusters from file
31+
Raw };
32+
enum struct OutputType { Clusters,
33+
Raw,
34+
DecodedClusters,
35+
Tracks };
36+
37+
/// create the workflow for TPC reconstruction
38+
framework::WorkflowSpec getWorkflow(bool propagateMC = true, int nLanes = 1, //
39+
std::string inputType = "digitizer", std::string outputType = "tracks" //
40+
);
41+
42+
} // end namespace RecoWorkflow
43+
} // end namespace TPC
44+
} // end namespace o2
45+
#endif //O2_TPC_RECOWORKFLOW_H

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

Lines changed: 20 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ namespace TPC
3939

4040
using MCLabelContainer = o2::dataformats::MCTruthContainer<o2::MCCompLabel>;
4141

42-
DataProcessorSpec getCATrackerSpec(bool processMC)
42+
DataProcessorSpec getCATrackerSpec(bool processMC, size_t fanIn)
4343
{
4444
constexpr static size_t NSectors = o2::TPC::Sector::MAXSECTOR;
4545
using ClusterGroupParser = o2::algorithm::ForwardParser<o2::TPC::ClusterGroupHeader>;
@@ -54,13 +54,15 @@ DataProcessorSpec getCATrackerSpec(bool processMC)
5454
std::unique_ptr<ClusterGroupParser> parser;
5555
std::unique_ptr<o2::TPC::TPCCATracking> tracker;
5656
int verbosity = 1;
57+
size_t nParallelInputs = 1;
5758
};
5859

59-
auto initFunction = [processMC](InitContext& ic) {
60+
auto initFunction = [processMC, fanIn](InitContext& ic) {
6061
auto options = ic.options().get<std::string>("tracker-options");
6162

6263
auto processAttributes = std::make_shared<ProcessAttributes>();
6364
{
65+
processAttributes->nParallelInputs = fanIn;
6466
auto& parser = processAttributes->parser;
6567
auto& tracker = processAttributes->tracker;
6668
parser = std::make_unique<ClusterGroupParser>();
@@ -81,7 +83,8 @@ DataProcessorSpec getCATrackerSpec(bool processMC)
8183
auto& mcInputs = processAttributes->mcInputs;
8284
if (processMC) {
8385
// we can later extend this to multiple inputs
84-
std::vector<std::string> inputLabels = { "mclblin" };
86+
std::vector<std::string> inputLabels(processAttributes->nParallelInputs);
87+
std::generate(inputLabels.begin(), inputLabels.end(), [counter = std::make_shared<int>(0)]() { return "mclblin" + std::to_string((*counter)++); });
8588
for (auto& inputLabel : inputLabels) {
8689
auto ref = pc.inputs().get(inputLabel);
8790
auto const* sectorHeader = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(ref);
@@ -115,8 +118,8 @@ DataProcessorSpec getCATrackerSpec(bool processMC)
115118
}
116119
}
117120

118-
// we can later extend this to multiple inputs
119-
std::vector<std::string> inputLabels = { "input" };
121+
std::vector<std::string> inputLabels(processAttributes->nParallelInputs);
122+
std::generate(inputLabels.begin(), inputLabels.end(), [counter = std::make_shared<int>(0)]() { return "input" + std::to_string((*counter)++); });
120123
auto& validInputs = processAttributes->validInputs;
121124
auto& inputs = processAttributes->inputs;
122125
for (auto& inputLabel : inputLabels) {
@@ -229,13 +232,17 @@ DataProcessorSpec getCATrackerSpec(bool processMC)
229232
return processingFct;
230233
};
231234

232-
auto createInputSpecs = [](bool makeMcInput) {
233-
std::vector<InputSpec> inputSpecs{
234-
InputSpec{ { "input" }, gDataOriginTPC, "CLUSTERNATIVE", 0, Lifetime::Timeframe },
235-
};
236-
if (makeMcInput) {
237-
constexpr o2::header::DataDescription datadesc("CLNATIVEMCLBL");
238-
inputSpecs.emplace_back(InputSpec{ "mclblin", gDataOriginTPC, datadesc, 0, Lifetime::Timeframe });
235+
auto createInputSpecs = [fanIn](bool makeMcInput) {
236+
std::vector<InputSpec> inputSpecs;
237+
for (size_t n = 0; n < fanIn; ++n) {
238+
std::string label = "input" + std::to_string(n);
239+
inputSpecs.emplace_back(InputSpec{ label, gDataOriginTPC, "CLUSTERNATIVE", n, Lifetime::Timeframe });
240+
241+
if (makeMcInput) {
242+
label = "mclblin" + std::to_string(n);
243+
constexpr o2::header::DataDescription datadesc("CLNATIVEMCLBL");
244+
inputSpecs.emplace_back(InputSpec{ label, gDataOriginTPC, datadesc, n, Lifetime::Timeframe });
245+
}
239246
}
240247
return std::move(inputSpecs);
241248
};
@@ -252,7 +259,7 @@ DataProcessorSpec getCATrackerSpec(bool processMC)
252259
return std::move(outputSpecs);
253260
};
254261

255-
return DataProcessorSpec{ "tracker", // process id
262+
return DataProcessorSpec{ "tpc-tracker", // process id
256263
{ createInputSpecs(processMC) },
257264
{ createOutputSpecs(false /*create onece writer process has been changed*/) },
258265
AlgorithmSpec(initFunction),

‎Detectors/TPC/workflow/src/CATrackerSpec.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ namespace TPC
2222

2323
/// create a processor spec
2424
/// read simulated TPC clusters from file and publish
25-
framework::DataProcessorSpec getCATrackerSpec(bool processMC = false);
25+
framework::DataProcessorSpec getCATrackerSpec(bool processMC = false, size_t fanIn = 1);
2626

2727
} // end namespace TPC
2828
} // end namespace o2

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

Lines changed: 38 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -40,33 +40,53 @@ using MCLabelContainer = o2::dataformats::MCTruthContainer<o2::MCCompLabel>;
4040
/// create a processor spec
4141
/// convert incoming TPC clusters to HW clusters
4242
/// Note: This processor does not touch the MC, see below
43-
DataProcessorSpec getClusterConverterSpec(bool sendMC)
43+
DataProcessorSpec getClusterConverterSpec(bool sendMC, int fanNumber)
4444
{
45-
auto initFunction = [sendMC](InitContext& ic) {
45+
std::string processorName = "tpc-cluster-converter";
46+
o2::header::DataHeader::SubSpecificationType fanSpec = 0;
47+
if (fanNumber < 0) {
48+
// only one instance
49+
fanNumber = 0;
50+
} else {
51+
// multiple instances, add number to name
52+
processorName += std::to_string(fanNumber);
53+
fanSpec = fanNumber;
54+
}
55+
56+
auto initFunction = [sendMC, fanSpec](InitContext& ic) {
4657
// there is nothing to init at the moment
4758
auto verbosity = 0;
4859

49-
auto processingFct = [verbosity, sendMC](ProcessingContext& pc) {
50-
// this will return a span of TPC clusters
51-
auto inClusters = pc.inputs().get<std::vector<o2::TPC::Cluster>>("clusterin");
52-
60+
auto processingFct = [verbosity, sendMC, fanSpec](ProcessingContext& pc) {
5361
// init the stacks for forwarding the sector header
5462
// FIXME check if there is functionality in the DPL to forward the stack
5563
// FIXME make one function
5664
o2::header::Stack rawHeaderStack;
5765
o2::header::Stack mcHeaderStack;
66+
o2::TPC::TPCSectorHeader const* sectorHeaderMC = nullptr;
5867
if (sendMC) {
59-
auto const* sectorHeader = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(pc.inputs().get("mclblin"));
60-
if (sectorHeader) {
61-
o2::header::Stack actual{ *sectorHeader };
68+
sectorHeaderMC = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(pc.inputs().get("mclblin"));
69+
if (sectorHeaderMC) {
70+
o2::header::Stack actual{ *sectorHeaderMC };
6271
std::swap(mcHeaderStack, actual);
72+
if (sectorHeaderMC->sector < 0) {
73+
pc.outputs().snapshot(OutputRef{ "mclblout", fanSpec, std::move(mcHeaderStack) }, fanSpec);
74+
}
6375
}
6476
}
6577
auto const* sectorHeader = DataRefUtils::getHeader<o2::TPC::TPCSectorHeader*>(pc.inputs().get("clusterin"));
6678
if (sectorHeader) {
6779
o2::header::Stack actual{ *sectorHeader };
6880
std::swap(rawHeaderStack, actual);
81+
if (sectorHeader->sector < 0) {
82+
pc.outputs().snapshot(OutputRef{ "clusterout", fanSpec, std::move(rawHeaderStack) }, fanSpec);
83+
return;
84+
}
6985
}
86+
assert(sectorHeaderMC == nullptr || sectorHeader->sector == sectorHeaderMC->sector);
87+
88+
// this will return a span of TPC clusters
89+
auto inClusters = pc.inputs().get<std::vector<o2::TPC::Cluster>>("clusterin");
7090
int nClusters = inClusters.size();
7191
LOG(INFO) << "got clusters from input: " << nClusters;
7292

@@ -118,7 +138,7 @@ DataProcessorSpec getClusterConverterSpec(bool sendMC)
118138
if (verbosity > 0) {
119139
LOG(INFO) << "allocating " << nTotalPages << " output page(s), " << nTotalPages * sizeof(ClusterHardwareContainer8kb);
120140
}
121-
auto outputPages = pc.outputs().make<ClusterHardwareContainer8kb>(OutputRef{ "clusterout", 0, std::move(rawHeaderStack) }, nTotalPages);
141+
auto outputPages = pc.outputs().make<ClusterHardwareContainer8kb>(OutputRef{ "clusterout", fanSpec, std::move(rawHeaderStack) }, nTotalPages);
122142

123143
auto outputPageIterator = outputPages.begin();
124144
unsigned mcoutIndex = 0;
@@ -166,7 +186,7 @@ DataProcessorSpec getClusterConverterSpec(bool sendMC)
166186
if (verbosity > 0) {
167187
LOG(INFO) << "writing MC labels for " << mcoutIndex << " cluster(s), index size " << mcout.getIndexedSize();
168188
}
169-
pc.outputs().snapshot(OutputRef{ "mclblout", 0, std::move(mcHeaderStack) }, mcout);
189+
pc.outputs().snapshot(OutputRef{ "mclblout", fanSpec, std::move(mcHeaderStack) }, mcout);
170190
}
171191
};
172192

@@ -182,32 +202,32 @@ DataProcessorSpec getClusterConverterSpec(bool sendMC)
182202
// as we do not expect any further sorting of clusters during the conversion, we do not
183203
// need to define the MC data at all, it is just routed directly to the final consumer.
184204
// Whether or not to have MC data is thus a feature of the initial producer.
185-
auto createInputSpecs = [](bool makeMcInput) {
205+
auto createInputSpecs = [fanSpec](bool makeMcInput) {
186206
std::vector<InputSpec> inputSpecs{
187-
InputSpec{ { "clusterin" }, gDataOriginTPC, "CLUSTERSIM", 0, Lifetime::Timeframe },
207+
InputSpec{ { "clusterin" }, gDataOriginTPC, "CLUSTERSIM", fanSpec, Lifetime::Timeframe },
188208
};
189209
if (makeMcInput) {
190210
// FIXME: define common data type specifiers
191211
constexpr o2::header::DataDescription datadesc("CLUSTERMCLBL");
192-
inputSpecs.emplace_back(InputSpec{ "mclblin", gDataOriginTPC, datadesc, 0, Lifetime::Timeframe });
212+
inputSpecs.emplace_back(InputSpec{ "mclblin", gDataOriginTPC, datadesc, fanSpec, Lifetime::Timeframe });
193213
}
194214
return std::move(inputSpecs);
195215
};
196216

197-
auto createOutputSpecs = [](bool makeMcOutput) {
217+
auto createOutputSpecs = [fanSpec](bool makeMcOutput) {
198218
std::vector<OutputSpec> outputSpecs{
199-
OutputSpec{ { "clusterout" }, gDataOriginTPC, "CLUSTERHW", 0, Lifetime::Timeframe },
219+
OutputSpec{ { "clusterout" }, gDataOriginTPC, "CLUSTERHW", fanSpec, Lifetime::Timeframe },
200220
};
201221
if (makeMcOutput) {
202222
OutputLabel label{ "mclblout" };
203223
// FIXME: define common data type specifiers
204224
constexpr o2::header::DataDescription datadesc("CLUSTERHWMCLBL");
205-
outputSpecs.emplace_back(label, gDataOriginTPC, datadesc, 0, Lifetime::Timeframe);
225+
outputSpecs.emplace_back(label, gDataOriginTPC, datadesc, fanSpec, Lifetime::Timeframe);
206226
}
207227
return std::move(outputSpecs);
208228
};
209229

210-
return DataProcessorSpec{ "converter",
230+
return DataProcessorSpec{ processorName,
211231
{ createInputSpecs(sendMC) },
212232
{ createOutputSpecs(sendMC) },
213233
AlgorithmSpec(initFunction) };

‎Detectors/TPC/workflow/src/ClusterConverterSpec.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ namespace TPC
2222

2323
/// create a processor spec
2424
/// read simulated TPC clusters from file and publish
25-
framework::DataProcessorSpec getClusterConverterSpec(bool sendMC);
25+
framework::DataProcessorSpec getClusterConverterSpec(bool sendMC, int fanNumber = -1);
2626

2727
} // end namespace TPC
2828
} // end namespace o2

0 commit comments

Comments
 (0)