Skip to content

Commit 0a66bda

Browse files
davidrohrdavidrohr
committed
GPU Workflow: start processing early asynchronously if possible, update inputs and outputcallback when DPL context of TF is available, wait for end of async processing, and ship data out via DPL
1 parent 9bd8464 commit 0a66bda

4 files changed

Lines changed: 89 additions & 25 deletions

File tree

‎GPU/Workflow/include/GPUWorkflow/GPUWorkflowSpec.h‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ class GPUO2InterfaceQA;
8585
struct GPUTrackingInOutPointers;
8686
struct GPUTrackingInOutZS;
8787
struct GPUInterfaceOutputs;
88+
struct GPUInterfaceInputUpdate;
8889
namespace gpurecoworkflow_internals
8990
{
9091
struct GPURecoWorkflowSpec_TPCZSBuffers;
@@ -167,8 +168,9 @@ class GPURecoWorkflowSpec : public o2::framework::Task
167168
void TerminateThreads();
168169
void handlePipelineEndOfStream(o2::framework::EndOfStreamContext& ec);
169170
void initPipeline(o2::framework::InitContext& ic);
170-
void enqueuePipelinedJob(GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, gpurecoworkflow_internals::GPURecoWorkflow_QueueObject* context);
171-
int runMain(o2::framework::ProcessingContext* pc, GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, int threadIndex);
171+
void enqueuePipelinedJob(GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, gpurecoworkflow_internals::GPURecoWorkflow_QueueObject* context, bool inputFinal);
172+
void finalizeInputPipelinedJob(GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, gpurecoworkflow_internals::GPURecoWorkflow_QueueObject* context);
173+
int runMain(o2::framework::ProcessingContext* pc, GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, int threadIndex = 0, GPUInterfaceInputUpdate* inputUpdateCallback = nullptr);
172174

173175
CompletionPolicyData* mPolicyData;
174176
std::function<bool(o2::framework::DataProcessingHeader::StartTime)> mPolicyOrder;

‎GPU/Workflow/src/GPUWorkflowInternal.h‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,13 +41,19 @@ struct GPURecoWorkflow_QueueObject {
4141
GPUTrackingInOutPointers ptrs;
4242
o2::framework::DataProcessingHeader::StartTime timeSliceId;
4343

44+
unsigned long mTFId;
45+
4446
bool jobSubmitted = false;
4547
bool jobFinished = false;
4648
int jobReturnValue = 0;
4749
std::mutex jobFinishedMutex;
4850
std::condition_variable jobFinishedNotify;
49-
GPUTrackingInOutPointers* jobPtrs;
50-
GPUInterfaceOutputs* jobOutputRegions;
51+
bool jobInputFinal = false;
52+
std::mutex jobInputFinalMutex;
53+
std::condition_variable jobInputFinalNotify;
54+
GPUTrackingInOutPointers* jobPtrs = nullptr;
55+
GPUInterfaceOutputs* jobOutputRegions = nullptr;
56+
std::unique_ptr<GPUInterfaceInputUpdate> jobInputUpdateCallback = nullptr;
5157
};
5258

5359
struct GPURecoWorkflowSpec_PipelineInternals {
@@ -75,6 +81,13 @@ struct GPURecoWorkflowSpec_PipelineInternals {
7581
bool pipelineSenderTerminating = false;
7682
std::mutex completionPolicyMutex;
7783
std::condition_variable completionPolicyNotify;
84+
85+
unsigned long mNTFReceived = 0;
86+
87+
bool mayInject = true;
88+
unsigned long mayInjectTFId = 0;
89+
std::mutex mayInjectMutex;
90+
std::condition_variable mayInjectCondition;
7891
};
7992

8093
} // namespace gpurecoworkflow_internals

‎GPU/Workflow/src/GPUWorkflowPipeline.cxx‎

Lines changed: 58 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ void GPURecoWorkflowSpec::RunWorkerThread(int id)
9191
context = workerContext.inputQueue.front();
9292
workerContext.inputQueue.pop();
9393
}
94-
context->jobReturnValue = runMain(nullptr, context->jobPtrs, context->jobOutputRegions, id);
94+
context->jobReturnValue = runMain(nullptr, context->jobPtrs, context->jobOutputRegions, id, context->jobInputUpdateCallback.get());
9595
{
9696
std::lock_guard lk(context->jobFinishedMutex);
9797
context->jobFinished = true;
@@ -100,11 +100,37 @@ void GPURecoWorkflowSpec::RunWorkerThread(int id)
100100
}
101101
}
102102

103-
void GPURecoWorkflowSpec::enqueuePipelinedJob(GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, GPURecoWorkflow_QueueObject* context)
103+
void GPURecoWorkflowSpec::enqueuePipelinedJob(GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, GPURecoWorkflow_QueueObject* context, bool inputFinal)
104104
{
105+
{
106+
std::unique_lock lk(mPipeline->mayInjectMutex);
107+
mPipeline->mayInjectCondition.wait(lk, [this, context]() { return mPipeline->mayInject && mPipeline->mayInjectTFId == context->mTFId; });
108+
mPipeline->mayInjectTFId++;
109+
mPipeline->mayInject = false;
110+
}
105111
context->jobSubmitted = true;
112+
context->jobInputFinal = inputFinal;
106113
context->jobPtrs = ptrs;
107114
context->jobOutputRegions = outputRegions;
115+
116+
context->jobInputUpdateCallback = std::make_unique<GPUInterfaceInputUpdate>();
117+
118+
if (!inputFinal) {
119+
context->jobInputUpdateCallback->callback = [context](GPUTrackingInOutPointers*& data, GPUInterfaceOutputs*& outputs) {
120+
std::unique_lock lk(context->jobInputFinalMutex);
121+
context->jobInputFinalNotify.wait(lk, [context]() { return context->jobInputFinal; });
122+
data = context->jobPtrs;
123+
outputs = context->jobOutputRegions;
124+
};
125+
}
126+
context->jobInputUpdateCallback->notifyCallback = [this]() {
127+
{
128+
std::lock_guard lk(mPipeline->mayInjectMutex);
129+
mPipeline->mayInject = true;
130+
}
131+
mPipeline->mayInjectCondition.notify_one();
132+
};
133+
108134
mNextThreadIndex = (mNextThreadIndex + 1) % 2;
109135

110136
{
@@ -114,6 +140,17 @@ void GPURecoWorkflowSpec::enqueuePipelinedJob(GPUTrackingInOutPointers* ptrs, GP
114140
mPipeline->workers[mNextThreadIndex].inputQueueNotify.notify_one();
115141
}
116142

143+
void GPURecoWorkflowSpec::finalizeInputPipelinedJob(GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, GPURecoWorkflow_QueueObject* context)
144+
{
145+
{
146+
std::lock_guard lk(context->jobInputFinalMutex);
147+
context->jobPtrs = ptrs;
148+
context->jobOutputRegions = outputRegions;
149+
context->jobInputFinal = true;
150+
}
151+
context->jobInputFinalNotify.notify_one();
152+
}
153+
117154
int GPURecoWorkflowSpec::handlePipeline(ProcessingContext& pc, GPUTrackingInOutPointers& ptrs, GPURecoWorkflowSpec_TPCZSBuffers& tpcZSmeta, o2::gpu::GPUTrackingInOutZS& tpcZS, std::unique_ptr<GPURecoWorkflow_QueueObject>& context)
118155
{
119156
auto* device = pc.services().get<RawDeviceService>().device();
@@ -246,9 +283,9 @@ void GPURecoWorkflowSpec::RunReceiveThread()
246283
break;
247284
}
248285

249-
auto o = std::make_unique<GPURecoWorkflow_QueueObject>();
250-
o->timeSliceId = m->timeSliceId;
251-
o->tfSettings = m->tfSettings;
286+
auto context = std::make_unique<GPURecoWorkflow_QueueObject>();
287+
context->timeSliceId = m->timeSliceId;
288+
context->tfSettings = m->tfSettings;
252289

253290
size_t regionOffset = 0;
254291
if (m->pointersTotal) {
@@ -263,29 +300,34 @@ void GPURecoWorkflowSpec::RunReceiveThread()
263300
}
264301
size_t ptrsCopied = 0;
265302
size_t* ptrBuffer = (size_t*)msg->GetData() + sizeof(pipelinePrepareMessage) / sizeof(size_t);
266-
o->tpcZSmeta.Pointers[0][0].resize(m->pointersTotal);
267-
o->tpcZSmeta.Sizes[0][0].resize(m->pointersTotal);
303+
context->tpcZSmeta.Pointers[0][0].resize(m->pointersTotal);
304+
context->tpcZSmeta.Sizes[0][0].resize(m->pointersTotal);
268305
for (unsigned int i = 0; i < GPUTrackingInOutZS::NSLICES; i++) {
269306
for (unsigned int j = 0; j < GPUTrackingInOutZS::NENDPOINTS; j++) {
270-
o->tpcZS.slice[i].count[j] = m->pointerCounts[i][j];
271-
for (unsigned int k = 0; k < o->tpcZS.slice[i].count[j]; k++) {
272-
o->tpcZSmeta.Pointers[0][0][ptrsCopied + k] = (void*)(ptrBuffer[ptrsCopied + k] + regionOffset);
273-
o->tpcZSmeta.Sizes[0][0][ptrsCopied + k] = ptrBuffer[m->pointersTotal + ptrsCopied + k];
307+
context->tpcZS.slice[i].count[j] = m->pointerCounts[i][j];
308+
for (unsigned int k = 0; k < context->tpcZS.slice[i].count[j]; k++) {
309+
context->tpcZSmeta.Pointers[0][0][ptrsCopied + k] = (void*)(ptrBuffer[ptrsCopied + k] + regionOffset);
310+
context->tpcZSmeta.Sizes[0][0][ptrsCopied + k] = ptrBuffer[m->pointersTotal + ptrsCopied + k];
274311
}
275-
o->tpcZS.slice[i].zsPtr[j] = o->tpcZSmeta.Pointers[0][0].data() + ptrsCopied;
276-
o->tpcZS.slice[i].nZSPtr[j] = o->tpcZSmeta.Sizes[0][0].data() + ptrsCopied;
277-
ptrsCopied += o->tpcZS.slice[i].count[j];
312+
context->tpcZS.slice[i].zsPtr[j] = context->tpcZSmeta.Pointers[0][0].data() + ptrsCopied;
313+
context->tpcZS.slice[i].nZSPtr[j] = context->tpcZSmeta.Sizes[0][0].data() + ptrsCopied;
314+
ptrsCopied += context->tpcZS.slice[i].count[j];
278315
}
279316
}
280-
o->ptrs.tpcZS = &o->tpcZS;
317+
context->ptrs.tpcZS = &context->tpcZS;
318+
context->ptrs.settingsTF = &context->tfSettings;
319+
context->mTFId = mPipeline->mNTFReceived;
320+
if (mPipeline->mNTFReceived++ >= mPipeline->workers.size()) { // Do not inject the first 2 TF, since we need a first round of calib updates from DPL before starting
321+
enqueuePipelinedJob(&context->ptrs, nullptr, context.get(), false);
322+
}
281323
{
282324
std::lock_guard lk(mPipeline->completionPolicyMutex);
283325
mPipeline->completionPolicyQueue.emplace(m->timeSliceId);
284326
}
285327
mPipeline->completionPolicyNotify.notify_one();
286328
{
287329
std::lock_guard lk(mPipeline->queueMutex);
288-
mPipeline->pipelineQueue.emplace(std::move(o));
330+
mPipeline->pipelineQueue.emplace(std::move(context));
289331
}
290332
mPipeline->queueNotify.notify_one();
291333
}

‎GPU/Workflow/src/GPUWorkflowSpec.cxx‎

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,9 @@ void GPURecoWorkflowSpec::init(InitContext& ic)
147147
throw std::runtime_error("GPU Event Display frontend could not be created!");
148148
}
149149
}
150+
if (mSpecConfig.enableDoublePipeline) {
151+
mConfig->configProcessing.doublePipeline = 1;
152+
}
150153

151154
mAutoSolenoidBz = mConfParam->solenoidBz == -1e6f;
152155
mAutoContinuousMaxTimeBin = mConfig->configGRP.continuousMaxTimeBin == -1;
@@ -487,20 +490,22 @@ void GPURecoWorkflowSpec::processInputs(ProcessingContext& pc, D& tpcZSmeta, E&
487490
}
488491
}
489492

490-
int GPURecoWorkflowSpec::runMain(o2::framework::ProcessingContext* pc, GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, int threadIndex)
493+
int GPURecoWorkflowSpec::runMain(o2::framework::ProcessingContext* pc, GPUTrackingInOutPointers* ptrs, GPUInterfaceOutputs* outputRegions, int threadIndex, GPUInterfaceInputUpdate* inputUpdateCallback)
491494
{
492495
int retVal = 0;
493496
if (mConfParam->dump < 2) {
494-
retVal = mGPUReco->RunTracking(ptrs, outputRegions, threadIndex);
497+
retVal = mGPUReco->RunTracking(ptrs, outputRegions, threadIndex, inputUpdateCallback);
495498

496499
if (retVal == 0 && mSpecConfig.runITSTracking) {
497500
retVal = runITSTracking(*pc);
498501
}
499502
}
500503

501504
cleanOldCalibsTPCPtrs(); // setting TPC calibration objects
502-
// if (!mTrackingCAO2Interface->getConfig().configInterface.outputToExternalBuffers) // TODO: Why was this needed for double-pipeline?
503-
mGPUReco->Clear(false, threadIndex); // clean non-output memory used by GPU Reconstruction
505+
506+
if (!mSpecConfig.enableDoublePipeline) { // TODO: Why is this needed for double-pipeline?
507+
mGPUReco->Clear(false, threadIndex); // clean non-output memory used by GPU Reconstruction
508+
}
504509
return retVal;
505510
}
506511

@@ -760,7 +765,9 @@ void GPURecoWorkflowSpec::run(ProcessingContext& pc)
760765
int retVal = 0;
761766
if (mSpecConfig.enableDoublePipeline) {
762767
if (!pipelineContext->jobSubmitted) {
763-
enqueuePipelinedJob(&ptrs, &outputRegions, pipelineContext.get());
768+
enqueuePipelinedJob(&ptrs, &outputRegions, pipelineContext.get(), true);
769+
} else {
770+
finalizeInputPipelinedJob(&ptrs, &outputRegions, pipelineContext.get());
764771
}
765772
std::unique_lock lk(pipelineContext->jobFinishedMutex);
766773
pipelineContext->jobFinishedNotify.wait(lk, [context = pipelineContext.get()]() { return context->jobFinished; });

0 commit comments

Comments
 (0)