From bede0bf3ed69b071881dc04eab2405582f6dd1cb Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Tue, 26 Sep 2023 10:59:58 +0200 Subject: [PATCH 1/2] DPL: fix race condition with the delivery of region events --- Framework/Core/src/DataProcessingDevice.cxx | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx index cf664a5e9fdeb..528770a980c09 100644 --- a/Framework/Core/src/DataProcessingDevice.cxx +++ b/Framework/Core/src/DataProcessingDevice.cxx @@ -957,9 +957,12 @@ void DataProcessingDevice::InitTask() LOG(detail) << "ptr: " << info.ptr; LOG(detail) << "size: " << info.size; LOG(detail) << "flags: " << info.flags; - context.expectedRegionCallbacks -= 1; + // Now we check for pending events with the mutex, + // so the lines below are atomic. pendingRegionInfos.push_back(info); - // We always want to handle these on the main loop + context.expectedRegionCallbacks -= 1; + // We always want to handle these on the main loop, + // so we awake it. ServiceRegistryRef ref{registry}; uv_async_send(ref.get().awakeMainThread); }); @@ -991,14 +994,21 @@ void DataProcessingDevice::InitTask() // We will get there. this->fillContext(mServiceRegistry.get(ServiceRegistry::globalDeviceSalt()), deviceContext); + auto hasPendingEvents = [&mutex = mRegionInfoMutex, &pendingRegionInfos = mPendingRegionInfos](DeviceContext& deviceContext) { + std::lock_guard lock(mutex); + return (pendingRegionInfos.empty() == false) || deviceContext.expectedRegionCallbacks > 0; + }; /// We now run an event loop also in InitTask. This is needed to: /// * Make sure region registration callbacks are invoked /// on the main thread. /// * Wait for enough callbacks to be delivered before moving to START - while (deviceContext.expectedRegionCallbacks > 0 && uv_run(state.loop, UV_RUN_ONCE)) { + while (hasPendingEvents(deviceContext)) { + // Wait for the callback to signal its done, so that we do not busy wait. + uv_run(state.loop, UV_RUN_ONCE); // Handle callbacks if any { std::lock_guard lock(mRegionInfoMutex); + // If there is no pending event, we can break the loop. handleRegionCallbacks(mServiceRegistry, mPendingRegionInfos); } } From dddc5170d293ab60b18748b97907a9ce35871b89 Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Tue, 26 Sep 2023 12:02:19 +0200 Subject: [PATCH 2/2] Update Framework/Core/src/DataProcessingDevice.cxx --- Framework/Core/src/DataProcessingDevice.cxx | 1 - 1 file changed, 1 deletion(-) diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx index 528770a980c09..e3c29035e9d21 100644 --- a/Framework/Core/src/DataProcessingDevice.cxx +++ b/Framework/Core/src/DataProcessingDevice.cxx @@ -1008,7 +1008,6 @@ void DataProcessingDevice::InitTask() // Handle callbacks if any { std::lock_guard lock(mRegionInfoMutex); - // If there is no pending event, we can break the loop. handleRegionCallbacks(mServiceRegistry, mPendingRegionInfos); } }