Skip to content

Commit cd1b7c7

Browse files
sawenzelclaude
andcommitted
Address review comments on the multithreaded TPC digitizer
This addresses the review of the multithreaded TPC digitizer device. - The workers copy the random rings of the first worker instead of filling their own. - The workers no longer flush the process-wide debug streamer; the device flushes it after the join. - The Digitizer serialises its calibration reads itself, so digitizeSector needs no mutex. - The CCDB handling, the calibration files and the common options of both devices are in TPCDigitizerConditions. - The device stops on a second timeframe, which the chunked digit files cannot hold. - Shorter comments in TPCSectorDigitization.h. https://its.cern.ch/jira/browse/O2-7309 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent ef82406 commit cd1b7c7

6 files changed

Lines changed: 257 additions & 215 deletions

File tree

‎Detectors/TPC/simulation/include/TPCSimulation/Digitizer.h‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,15 @@ class Digitizer
154154
/// Set the timeframe identifier that enters the sector seeding
155155
void setSeedingTimeframe(ULong64_t timeframe) { mSeedingTimeframe = timeframe; }
156156

157+
/// Fill the random rings of the sector seeding, if not done yet
158+
void initSectorSeeding();
159+
160+
/// Use copies of the sector-seeding random rings of another digitizer
161+
void setSectorSeedingFrom(const Digitizer& other);
162+
163+
/// Whether flush() also flushes the process-wide debug streamer
164+
void setFlushDebugStreamer(bool flush) { mFlushDebugStreamer = flush; }
165+
157166
private:
158167
ElectronTransport& electronTransport();
159168
GEMAmplification& gemAmplification();
@@ -165,6 +174,7 @@ class Digitizer
165174
ULong64_t mSeed = 0; //!< Base seed for sector seeding
166175
ULong64_t mSeedingTimeframe = 0; //!< Timeframe identifier for sector seeding
167176
bool mUseSectorSeeding = false; //!< Whether sector seeding is used
177+
bool mFlushDebugStreamer = true; //!< Whether flush() flushes the debug streamer
168178

169179
DigitContainer mDigitContainer; ///< Container for the Digits
170180
std::vector<float> mSignalArray; //!< Workspace for the shaped signal of one electron

‎Detectors/TPC/simulation/src/Digitizer.cxx‎

Lines changed: 39 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131

3232
#include <fairlogger/Logger.h>
3333
#include <TRandom3.h>
34+
#include <mutex>
3435

3536
ClassImp(o2::tpc::Digitizer);
3637

@@ -42,6 +43,13 @@ Digitizer::Digitizer() = default;
4243

4344
namespace
4445
{
46+
/// serialises the calls that read calibrations through the CDBInterface, which is not thread safe
47+
std::mutex& calibrationMutex()
48+
{
49+
static std::mutex mutex;
50+
return mutex;
51+
}
52+
4553
// splitmix64 finaliser, to derive well separated seeds from the base seed
4654
ULong64_t mixSeed(ULong64_t x)
4755
{
@@ -80,17 +88,35 @@ void Digitizer::setUseSectorSeeding(ULong64_t seed)
8088
mSeed = seed;
8189
}
8290

91+
void Digitizer::initSectorSeeding()
92+
{
93+
if (mUseSectorSeeding && !mGEMAmplification) {
94+
// the rings are filled once, in the construction order of the singletons
95+
TRandom3 rng(toTRandom3Seed(mixSeed(mSeed)));
96+
mGEMAmplification = std::make_unique<GEMAmplification>(&rng);
97+
mElectronTransport = std::make_unique<ElectronTransport>(&rng);
98+
mSAMPAProcessing = std::make_unique<SAMPAProcessing>(&rng);
99+
mDigitContainer.setSAMPAProcessing(mSAMPAProcessing.get());
100+
}
101+
}
102+
103+
void Digitizer::setSectorSeedingFrom(const Digitizer& other)
104+
{
105+
mUseSectorSeeding = other.mUseSectorSeeding;
106+
mSeed = other.mSeed;
107+
if (other.mGEMAmplification) {
108+
mGEMAmplification = std::make_unique<GEMAmplification>(*other.mGEMAmplification);
109+
mElectronTransport = std::make_unique<ElectronTransport>(*other.mElectronTransport);
110+
mSAMPAProcessing = std::make_unique<SAMPAProcessing>(*other.mSAMPAProcessing);
111+
mDigitContainer.setSAMPAProcessing(mSAMPAProcessing.get());
112+
}
113+
}
114+
83115
void Digitizer::init()
84116
{
117+
std::lock_guard<std::mutex> lock(calibrationMutex());
85118
if (mUseSectorSeeding) {
86-
if (!mGEMAmplification) {
87-
// the rings are filled once, in the construction order of the singletons
88-
TRandom3 rng(toTRandom3Seed(mixSeed(mSeed)));
89-
mGEMAmplification = std::make_unique<GEMAmplification>(&rng);
90-
mElectronTransport = std::make_unique<ElectronTransport>(&rng);
91-
mSAMPAProcessing = std::make_unique<SAMPAProcessing>(&rng);
92-
mDigitContainer.setSAMPAProcessing(mSAMPAProcessing.get());
93-
}
119+
initSectorSeeding();
94120
const ULong64_t sectorSeed = mixSeed(mSeed ^ mixSeed(static_cast<ULong64_t>(mSector) + 1));
95121
TRandom3 rng(toTRandom3Seed(mixSeed(sectorSeed ^ mixSeed(mSeedingTimeframe))));
96122
mGEMAmplification->setRandomRingPositions(rng);
@@ -230,7 +256,7 @@ void Digitizer::flush(std::vector<o2::tpc::Digit>& digits,
230256
SAMPAProcessing& sampaProcessing = this->sampaProcessing();
231257
mDigitContainer.fillOutputContainer(digits, labels, commonModeOutput, mSector, sampaProcessing.getTimeBinFromTime(mEventTime - mOutputDigitTimeOffset), mIsContinuous, finalFlush);
232258
// flushing debug output to file
233-
if (((finalFlush && mIsContinuous) || (!mIsContinuous)) && mSpaceCharge) {
259+
if (mFlushDebugStreamer && ((finalFlush && mIsContinuous) || (!mIsContinuous)) && mSpaceCharge) {
234260
o2::utils::DebugStreamer::instance()->flush();
235261
}
236262
}
@@ -289,7 +315,10 @@ void Digitizer::setStartTime(double time)
289315
// note that negative times w.r.t start of timeframe/data-taking == mOutputDigitTimeOffset
290316
// will yield the 0-th bin (due to casting logic in sampaProcessing)
291317
SAMPAProcessing& sampaProcessing = this->sampaProcessing();
292-
sampaProcessing.updateParameters(mVDrift);
318+
{
319+
std::lock_guard<std::mutex> lock(calibrationMutex());
320+
sampaProcessing.updateParameters(mVDrift);
321+
}
293322
const auto timediff = time - mOutputDigitTimeOffset;
294323
const auto starttimebin = sampaProcessing.getTimeBinFromTime(timediff);
295324
mDigitContainer.setStartTime(starttimebin);

‎Steer/DigitizerWorkflow/src/TPCDigitizerMTSpec.cxx‎

Lines changed: 34 additions & 94 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
#include "TPCCalibration/VDriftHelper.h"
3535
#include "TPCSimulation/Digitizer.h"
3636
#include "TPCSpaceCharge/SpaceCharge.h"
37+
#include "CommonUtils/DebugStreamer.h"
3738
#include <TChain.h>
3839
#include <TFile.h>
3940
#include <TROOT.h>
@@ -45,7 +46,6 @@
4546
#include <exception>
4647
#include <filesystem>
4748
#include <memory>
48-
#include <mutex>
4949
#include <optional>
5050
#include <string>
5151
#include <thread>
@@ -80,7 +80,7 @@ struct DigitizerWorker {
8080
std::string fileName() const { return "tpc_driftime_digits_lane" + std::to_string(id) + ".root"; }
8181

8282
/// Digitize one sector and write it to the worker's digit file, in the format of the chunked writer of the lane spec
83-
void processSector(const steer::DigitizationContext& context, int sector, ULong64_t timeframe, bool withMCTruth, std::mutex& cdbMutex)
83+
void processSector(const steer::DigitizationContext& context, int sector, ULong64_t timeframe, bool withMCTruth)
8484
{
8585
std::unique_ptr<TFile> file(TFile::Open(fileName().c_str(), "UPDATE"));
8686
if (!file || file->IsZombie()) {
@@ -102,7 +102,7 @@ struct DigitizerWorker {
102102
}
103103
fill("TPCCommonMode_" + std::to_string(sector), &buffers.commonMode);
104104
};
105-
digitizeSector(digitizer, context, chains, sector, timeframe, buffers, onFlush, &cdbMutex);
105+
digitizeSector(digitizer, context, chains, sector, timeframe, buffers, onFlush);
106106
tree->SetEntries(flushCounter);
107107
file->Write("", TObject::kOverwrite);
108108
file->Close();
@@ -121,12 +121,7 @@ class TPCDPLDigitizerMTTask : public o2::base::BaseDPLDigitizer
121121

122122
mWithMCTruth = o2::conf::DigiParams::Instance().mctruth;
123123
const auto triggeredMode = ic.options().get<bool>("TPCtriggered");
124-
mRecalcDistortions = !(ic.options().get<bool>("do-not-recalculate-distortions"));
125-
SC::setNThreads(ic.options().get<int>("n-threads-distortions"));
126-
mUseCalibrationsFromCCDB = ic.options().get<bool>("TPCuseCCDB");
127-
mMeanLumiDistortions = ic.options().get<float>("meanLumiDistortions");
128-
mMeanLumiDistortionsDerivative = ic.options().get<float>("meanLumiDistortionsDerivative");
129-
LOG(info) << "TPC calibrations from CCDB: " << mUseCalibrationsFromCCDB;
124+
mConditions.init(ic, mDistortionType);
130125

131126
// the workers share no random numbers: each digitizer seeds every sector from DigiParams.seed
132127
const auto seed = o2::conf::DigiParams::Instance().seed;
@@ -142,6 +137,8 @@ class TPCDPLDigitizerMTTask : public o2::base::BaseDPLDigitizer
142137
worker->digitizer.setContinuousReadout(!triggeredMode);
143138
worker->digitizer.setDistortionScaleType(mDistortionType);
144139
worker->digitizer.setUseSectorSeeding(sectorSeed);
140+
// the main thread flushes the process-wide debug streamer after the workers
141+
worker->digitizer.setFlushDebugStreamer(false);
145142
}
146143

147144
// the merger picks up every lane file in the directory, so files of an earlier run must not survive
@@ -157,80 +154,33 @@ class TPCDPLDigitizerMTTask : public o2::base::BaseDPLDigitizer
157154

158155
void finaliseCCDB(framework::ConcreteDataMatcher& matcher, void* obj)
159156
{
160-
if (mTPCVDriftHelper.accountCCDBInputs(matcher, obj)) {
161-
return;
162-
}
163157
// the maps are set on the first digitizer; the others share them once they are final
164-
auto& digitizer = mWorkers[0]->digitizer;
165-
if (matcher == ConcreteDataMatcher(o2::header::gDataOriginTPC, "TPCDIST", 0)) {
166-
LOGP(info, "Updating distortion map");
167-
digitizer.setUseSCDistortions(static_cast<SC*>(obj));
168-
if (mMeanLumiDistortions >= 0) {
169-
digitizer.setMeanLumiDistortions(mMeanLumiDistortions);
170-
}
171-
}
172-
if (matcher == ConcreteDataMatcher(o2::header::gDataOriginTPC, "TPCDISTDERIV", 0)) {
173-
LOGP(info, "Updating reference distortion map");
174-
digitizer.setSCDistortionsDerivative(static_cast<SC*>(obj));
175-
if (mMeanLumiDistortionsDerivative >= 0) {
176-
digitizer.setMeanLumiDistortionsDerivative(mMeanLumiDistortionsDerivative);
177-
}
178-
}
179-
if (matcher == ConcreteDataMatcher(o2::header::gDataOriginTPC, "TPCGASPARAM", 0)) {
180-
LOGP(info, "TPC gas param updated");
181-
ParameterGas::Instance().printKeyValues(true, true);
182-
}
183-
if (matcher == ConcreteDataMatcher(o2::header::gDataOriginTPC, "TPCGEMPARAM", 0)) {
184-
LOGP(info, "TPC GEM param updated");
185-
ParameterGEM::Instance().printKeyValues(true, true);
186-
}
158+
mConditions.finaliseCCDB(matcher, obj, mWorkers[0]->digitizer);
187159
}
188160

189161
void run(framework::ProcessingContext& pc)
190162
{
191163
LOG(info) << "Processing multithreaded TPC digitization";
192164

193-
auto& cdb = o2::tpc::CDBInterface::instance();
194-
cdb.setUseDefaults(!mUseCalibrationsFromCCDB);
195-
// whatever are global settings for CCDB usage, we have to extract the TPC vdrift from CCDB for anchored simulations
196-
mTPCVDriftHelper.extractCCDBInputs(pc);
197-
pc.inputs().get<ParameterGas*>("gasparam");
198-
pc.inputs().get<ParameterGEM*>("gemparam");
199165
auto& primary = mWorkers[0]->digitizer;
200-
if (mDistortionType) {
201-
pc.inputs().get<SC*>("tpcdistortions");
202-
if (mDistortionType == 2) {
203-
pc.inputs().get<SC*>("tpcdistortionsderiv");
204-
primary.setLumiScaleFactor();
205-
if (mRecalcDistortions) {
206-
// done once, with the threads of SpaceCharge, for all workers
207-
primary.recalculateDistortions();
208-
}
166+
mConditions.update(pc, primary, [this](float vdrift, float tdriftOffset) {
167+
for (auto& worker : mWorkers) {
168+
worker->digitizer.setVDrift(vdrift);
169+
worker->digitizer.setTDriftOffset(tdriftOffset);
209170
}
171+
});
172+
if (mDistortionType) {
210173
for (size_t i = 1; i < mWorkers.size(); ++i) {
211174
mWorkers[i]->digitizer.setSCDistortionsFrom(primary);
212175
}
213176
}
214-
215-
if (mTPCVDriftHelper.isUpdated()) {
216-
const auto& vd = mTPCVDriftHelper.getVDriftObject();
217-
LOGP(info, "Updating TPC fast transform map with new VDrift factor of {} wrt reference {} and DriftTimeOffset correction {} wrt {} from source {}",
218-
vd.corrFact, vd.refVDrift, vd.timeOffsetCorr, vd.refTimeOffset, mTPCVDriftHelper.getSourceName());
219-
for (auto& worker : mWorkers) {
220-
worker->digitizer.setVDrift(vd.getVDrift());
221-
worker->digitizer.setTDriftOffset(vd.getTimeOffset());
177+
if (!mSeedingShared) {
178+
// the rings depend only on the seed, so they are filled once and copied
179+
primary.initSectorSeeding();
180+
for (size_t i = 1; i < mWorkers.size(); ++i) {
181+
mWorkers[i]->digitizer.setSectorSeedingFrom(primary);
222182
}
223-
mTPCVDriftHelper.acknowledgeUpdate();
224-
}
225-
226-
if (std::filesystem::exists("ThresholdMap.root")) {
227-
LOG(info) << "TPC: Using zero suppression map from 'ThresholdMap.root'";
228-
cdb.setThresholdMapFromFile("ThresholdMap.root");
229-
}
230-
231-
if (std::filesystem::exists("GainMap.root")) {
232-
LOG(info) << "TPC: Using gain map from 'GainMap.root'";
233-
cdb.setGainMapFromFile("GainMap.root");
183+
mSeedingShared = true;
234184
}
235185

236186
// the sectors of this timeframe; all inputs carry the same collision context
@@ -266,6 +216,11 @@ class TPCDPLDigitizerMTTask : public o2::base::BaseDPLDigitizer
266216
if (!contextRef) {
267217
return;
268218
}
219+
// the chunked digit files and the dynamic sector scheduling hold one timeframe per process
220+
if (mProcessedTimeframe) {
221+
LOG(fatal) << "TPC: the multithreaded digitizer supports only one timeframe per process";
222+
}
223+
mProcessedTimeframe = true;
269224
auto context = pc.inputs().get<o2::steer::DigitizationContext*>(*contextRef);
270225
const ULong64_t timeframe = DataRefUtils::getHeader<o2::header::DataHeader*>(*contextRef)->tfCounter;
271226
LOG(info) << "TPC: Processing " << sectors.size() << " sectors with " << context->getEventRecords().size() << " collisions";
@@ -278,16 +233,15 @@ class TPCDPLDigitizerMTTask : public o2::base::BaseDPLDigitizer
278233
context->initSimChains(o2::detectors::DetID::TPC, mWorkers[i]->chains);
279234
}
280235

236+
// the sectors go to the workers dynamically
281237
std::atomic<size_t> nextSector{0};
282-
// serialises CDBInterface access; the CCDB timestamp must not change while the workers run
283-
std::mutex cdbMutex;
284238
std::vector<std::thread> threads;
285239
for (size_t i = 0; i < nWorkers; ++i) {
286240
threads.emplace_back([&, worker = mWorkers[i].get()]() {
287241
try {
288242
for (size_t is = nextSector++; is < sectors.size(); is = nextSector++) {
289243
LOG(info) << "TPC: Worker " << worker->id << " processing sector " << sectors[is];
290-
worker->processSector(*context, sectors[is], timeframe, mWithMCTruth, cdbMutex);
244+
worker->processSector(*context, sectors[is], timeframe, mWithMCTruth);
291245
}
292246
} catch (...) {
293247
worker->error = std::current_exception();
@@ -298,6 +252,9 @@ class TPCDPLDigitizerMTTask : public o2::base::BaseDPLDigitizer
298252
for (auto& thread : threads) {
299253
thread.join();
300254
}
255+
if (auto* streamer = o2::utils::DebugStreamer::instance()) {
256+
streamer->flush();
257+
}
301258
for (size_t i = 0; i < nWorkers; ++i) {
302259
if (mWorkers[i]->error) {
303260
std::rethrow_exception(mWorkers[i]->error);
@@ -310,15 +267,13 @@ class TPCDPLDigitizerMTTask : public o2::base::BaseDPLDigitizer
310267

311268
private:
312269
std::vector<std::unique_ptr<DigitizerWorker>> mWorkers;
313-
o2::tpc::VDriftHelper mTPCVDriftHelper{};
270+
TPCDigitizerConditions mConditions;
271+
bool mSeedingShared = false; ///< whether the workers got the random rings of the first one
272+
bool mProcessedTimeframe = false; ///< whether a timeframe was digitized
314273
int mNThreads = 1;
315274
int mDistortionType = 0;
316275
bool mWriteGRP = true;
317276
bool mWithMCTruth = true;
318-
bool mUseCalibrationsFromCCDB = false;
319-
bool mRecalcDistortions = false;
320-
float mMeanLumiDistortions = -1;
321-
float mMeanLumiDistortionsDerivative = -1;
322277
};
323278

324279
o2::framework::DataProcessorSpec getTPCDigitizerMTSpec(int nThreads, std::vector<int> const& sectors, int distortionType)
@@ -327,15 +282,7 @@ o2::framework::DataProcessorSpec getTPCDigitizerMTSpec(int nThreads, std::vector
327282
for (auto sector : sectors) {
328283
inputs.emplace_back("collisioncontext" + std::to_string(sector), "SIM", "COLLISIONCONTEXT", static_cast<o2::header::DataHeader::SubSpecificationType>(sector), Lifetime::Timeframe);
329284
}
330-
o2::tpc::VDriftHelper::requestCCDBInputs(inputs);
331-
inputs.emplace_back("gasparam", o2::header::gDataOriginTPC, "TPCGASPARAM", 0, Lifetime::Condition, ccdbParamSpec(o2::tpc::CDBTypeMap.at(o2::tpc::CDBType::ParGas)));
332-
inputs.emplace_back("gemparam", o2::header::gDataOriginTPC, "TPCGEMPARAM", 0, Lifetime::Condition, ccdbParamSpec(o2::tpc::CDBTypeMap.at(o2::tpc::CDBType::ParGEM)));
333-
if (distortionType) {
334-
inputs.emplace_back("tpcdistortions", o2::header::gDataOriginTPC, "TPCDIST", 0, Lifetime::Condition, ccdbParamSpec(o2::tpc::CDBTypeMap.at(o2::tpc::CDBType::DistortionMapMC), {}, 1)); // time-dependent
335-
if (distortionType == 2) {
336-
inputs.emplace_back("tpcdistortionsderiv", o2::header::gDataOriginTPC, "TPCDISTDERIV", 0, Lifetime::Condition, ccdbParamSpec(o2::tpc::CDBTypeMap.at(o2::tpc::CDBType::DistortionMapDerivMC), {}, 1)); // time-dependent
337-
}
338-
}
285+
TPCDigitizerConditions::addInputs(inputs, distortionType);
339286
std::vector<OutputSpec> outputs{{"TPC", "ROMode", 0, Lifetime::Timeframe}};
340287

341288
// the name must not match "TPCDigitizer.*", whose completion policy would process the sectors one by one
@@ -344,14 +291,7 @@ o2::framework::DataProcessorSpec getTPCDigitizerMTSpec(int nThreads, std::vector
344291
inputs,
345292
outputs,
346293
AlgorithmSpec{adaptFromTask<TPCDPLDigitizerMTTask>(nThreads, distortionType)},
347-
Options{
348-
{"TPCtriggered", VariantType::Bool, false, {"Impose triggered RO mode (default: continuous)"}},
349-
{"TPCuseCCDB", VariantType::Bool, false, {"true: load calibrations from CCDB; false: use random calibratoins"}},
350-
{"meanLumiDistortions", VariantType::Float, -1.f, {"override lumi of distortion object if >=0"}},
351-
{"meanLumiDistortionsDerivative", VariantType::Float, -1.f, {"override lumi of derivative distortion object if >=0"}},
352-
{"do-not-recalculate-distortions", VariantType::Bool, false, {"Do not recalculate the distortions"}},
353-
{"n-threads-distortions", VariantType::Int, 4, {"Number of threads used for the calculation of the distortions"}},
354-
}};
294+
TPCDigitizerConditions::options()};
355295
}
356296

357297
} // namespace tpc

0 commit comments

Comments
 (0)