From e9c1c0f0647d397b0481bde53a3e2082a1df8538 Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Sun, 8 Nov 2026 17:07:47 +0100 Subject: [PATCH 1/2] DPL Analysis: send a freshly fetched CCDB object once On a cache miss the analysis CCDB fetcher adopted the new object with CacheStrategy::Always, which caches and sends it, and then sent it a second time from AnalysisCCDBHelpers via adoptFromCache, which already sends every object referenced by the table once per timeframe. Add CacheStrategy::CacheOnly, which caches the message without sending it, and use it for the analysis fetcher, so that adoptFromCache is the only place its objects are sent from. The non-analysis CCDB fetcher keeps CacheStrategy::Always. --- Framework/CCDBSupport/src/CCDBFetcherHelper.cxx | 4 ++-- Framework/CCDBSupport/src/CCDBHelpers.cxx | 5 +++-- Framework/CCDBSupport/src/CCDBHelpers.h | 4 +++- Framework/Core/include/Framework/DataAllocator.h | 9 +++++++-- 4 files changed, 15 insertions(+), 7 deletions(-) diff --git a/Framework/CCDBSupport/src/CCDBFetcherHelper.cxx b/Framework/CCDBSupport/src/CCDBFetcherHelper.cxx index e4a9556c59b17..2479db4300850 100644 --- a/Framework/CCDBSupport/src/CCDBFetcherHelper.cxx +++ b/Framework/CCDBSupport/src/CCDBFetcherHelper.cxx @@ -256,7 +256,7 @@ auto CCDBFetcherHelper::populateCacheWith(std::shared_ptr con helper->totalFetchedBytes += size; helper->totalRequestedBytes += size; api.appendFlatHeader(v, headers); - auto cacheId = CCDBHelpers::adoptAndReplaceCachedMessage(allocator, helper->mapURL2DPLCache, path, output, std::move(v), header::gSerializationMethodCCDB); + auto cacheId = CCDBHelpers::adoptAndReplaceCachedMessage(allocator, helper->mapURL2DPLCache, path, output, std::move(v), header::gSerializationMethodCCDB, DataAllocator::CacheStrategy::CacheOnly); helper->mapURL2DPLCache[path] = cacheId; responses.emplace_back(Response{.id = cacheId, .size = size, .request = nullptr}); O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "Caching %{public}s for %{public}s (DPL id %" PRIu64 ", size %zu)", path.data(), headers["ETag"].data(), cacheId.value, size); @@ -274,7 +274,7 @@ auto CCDBFetcherHelper::populateCacheWith(std::shared_ptr con helper->totalFetchedBytes += size; helper->totalRequestedBytes += size; api.appendFlatHeader(v, headers); - auto cacheId = CCDBHelpers::adoptAndReplaceCachedMessage(allocator, helper->mapURL2DPLCache, path, output, std::move(v), header::gSerializationMethodCCDB); + auto cacheId = CCDBHelpers::adoptAndReplaceCachedMessage(allocator, helper->mapURL2DPLCache, path, output, std::move(v), header::gSerializationMethodCCDB, DataAllocator::CacheStrategy::CacheOnly); helper->mapURL2DPLCache[path] = cacheId; responses.emplace_back(Response{.id = cacheId, .size = size, .request = nullptr}); O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "Caching %{public}s for %{public}s (DPL id %" PRIu64 ")", path.data(), headers["ETag"].data(), cacheId.value); diff --git a/Framework/CCDBSupport/src/CCDBHelpers.cxx b/Framework/CCDBSupport/src/CCDBHelpers.cxx index 47ba6f00953cf..dba5d30dd370b 100644 --- a/Framework/CCDBSupport/src/CCDBHelpers.cxx +++ b/Framework/CCDBSupport/src/CCDBHelpers.cxx @@ -255,10 +255,11 @@ DataAllocator::CacheId CCDBHelpers::adoptAndReplaceCachedMessage( std::string const& path, Output const& output, o2::pmr::vector&& v, - o2::header::SerializationMethod method) + o2::header::SerializationMethod method, + DataAllocator::CacheStrategy strategy) { auto oldIt = cache.find(path); - auto cacheId = allocator.adoptContainer(output, std::move(v), DataAllocator::CacheStrategy::Always, method); + auto cacheId = allocator.adoptContainer(output, std::move(v), strategy, method); if (oldIt != cache.end()) { allocator.pruneFromCache(oldIt->second); } diff --git a/Framework/CCDBSupport/src/CCDBHelpers.h b/Framework/CCDBSupport/src/CCDBHelpers.h index ad67e2c64558e..e69bc6867e81d 100644 --- a/Framework/CCDBSupport/src/CCDBHelpers.h +++ b/Framework/CCDBSupport/src/CCDBHelpers.h @@ -39,6 +39,7 @@ struct CCDBHelpers { /// @a output DPL Output matcher /// @a v freshly-fetched CCDB payload; consumed by the call, leaving @a v empty /// @a method serialization-method tag written into the message header + /// @a strategy CacheOnly to cache the message without sending it /// @return the new CacheId; the caller must record it in its map static DataAllocator::CacheId adoptAndReplaceCachedMessage( DataAllocator& allocator, @@ -46,7 +47,8 @@ struct CCDBHelpers { std::string const& path, Output const& output, o2::pmr::vector&& v, - o2::header::SerializationMethod method); + o2::header::SerializationMethod method, + DataAllocator::CacheStrategy strategy = DataAllocator::CacheStrategy::Always); }; } // namespace o2::framework diff --git a/Framework/Core/include/Framework/DataAllocator.h b/Framework/Core/include/Framework/DataAllocator.h index 43dda80eba3a4..48d571031c72a 100644 --- a/Framework/Core/include/Framework/DataAllocator.h +++ b/Framework/Core/include/Framework/DataAllocator.h @@ -528,7 +528,9 @@ class DataAllocator enum struct CacheStrategy : int { Never = 0, - Always = 1 + Always = 1, + /// Cache the message without sending it. Use adoptFromCache to send it. + CacheOnly = 2 }; template @@ -628,7 +630,7 @@ DataAllocator::CacheId DataAllocator::adoptContainer(const Output& spec, Contain ); CacheId cacheId{0, 0, 0}; // - if (cache == CacheStrategy::Always) { + if (cache != CacheStrategy::Never) { // The message will be shallow cloned in the cache. Since the // clone is indistinguishable from the original, we can keep sending // the original. @@ -637,6 +639,9 @@ DataAllocator::CacheId DataAllocator::adoptContainer(const Output& spec, Contain cacheId.handle = meta.fHandle; cacheId.segment = meta.fSegmentId; } + if (cache == CacheStrategy::CacheOnly) { + return cacheId; + } context.add(std::move(headerMessage), std::move(payloadMessage), routeIndex); return cacheId; From b1c1e3b4535d7b0589569c4fd73b03a7d60bbc56 Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Sun, 8 Nov 2026 17:07:48 +0100 Subject: [PATCH 2/2] DPL Analysis: send CCDB tables together with the objects they refer to The analysis CCDB fetcher sent each CCDB table through the arrow backend and the CCDB objects it refers to, which keep them alive in the consumer, through the message backend. The two backends send separately, the arrow one first, and the consumer's DataRelayer completes the input on its first part. When every other input of the consumer was already there, as for the first timeframe, the slot was processed with the table alone and the objects then opened a new slot that could never complete, which was dropped with "Dropping incomplete AOD/GLOCCDBOBJ/1 ...". Serialise the table in the fetcher and send it through the message backend as well, followed by the objects. They then leave in one multipart, which the consumer relays as a whole before checking for completion, so they always land in the same slot with the table as the first part. The table is adopted with adoptContainer, like the objects, since messages created via snapshot are scheduled and would be sent after them. Its header and the arrow byte counters are set as the arrow backend does, so the rate limiting accounts for it as before. The alternatives were ruled out: - A separate input for the objects changes the topology of every analysis workflow using CCDB tables. - Not sending the objects at all and relying on the fetcher's cache would free an object still in use downstream when it is replaced, or when the fetcher exits before its consumers. - Swapping the order of the backends would make an object, when it arrives first, complete the input and be read as the table. Add a DataRelayer test for parts relayed from one multipart, which complete together with the table first, against parts relayed with a completion check in between, which reproduce the drop. --- .../CCDBSupport/src/AnalysisCCDBHelpers.cxx | 29 ++++++- Framework/Core/test/test_DataRelayer.cxx | 85 +++++++++++++++++++ 2 files changed, 110 insertions(+), 4 deletions(-) diff --git a/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx b/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx index 5bd21ce6bdf3f..aa1c35cc73b17 100644 --- a/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx +++ b/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx @@ -11,6 +11,7 @@ #include "AnalysisCCDBHelpers.h" #include "CCDBFetcherHelper.h" +#include "Framework/ArrowContext.h" #include "Framework/ArrowTypes.h" #include "Framework/DataProcessingStats.h" #include "Framework/DeviceSpec.h" @@ -25,6 +26,8 @@ #include "Framework/ConfigParamsHelper.h" #include #include +#include +#include #include #include #include @@ -161,7 +164,7 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) std::unordered_map bindings; fillValidRoutes(*helper, spec.outputs, bindings); - return adaptStateless([schemas, bindings, helper, allbuilders, pathTables](InputRecord& inputs, DataTakingContext& dtc, DataAllocator& allocator, TimingInfo& timingInfo, DataProcessingStats& stats) { + return adaptStateless([schemas, bindings, helper, allbuilders, pathTables](InputRecord& inputs, DataTakingContext& dtc, DataAllocator& allocator, ArrowContext& arrowContext, TimingInfo& timingInfo, DataProcessingStats& stats) { O2_SIGNPOST_ID_GENERATE(sid, ccdb); O2_SIGNPOST_START(ccdb, sid, "fetchFromAnalysisCCDB", "Fetching CCDB objects for analysis%" PRIu64, (uint64_t)timingInfo.timeslice); std::ranges::for_each(allbuilders, [](auto& builder) { builder.second->Reset(); }); @@ -232,6 +235,7 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) } std::vector lastIds(numBuilders, DataAllocator::CacheId{.value = -1, .handle = -1, .segment = -1}); + std::vector referenced; // Rows sharing a uniformity value resolve to the same objects, so the query is // issued once per distinct value and the resulting handles are repeated for the @@ -323,7 +327,7 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) auto& lastId = lastIds[bi]; if (response.id.value != lastId.value) { lastId.value = response.id.value; - allocator.adoptFromCache(output, response.id, header::gSerializationMethodCCDB); + referenced.push_back(response.id); } result &= builder.second->Append(); auto* value_builder = dynamic_cast(builder.second->value_builder()); @@ -340,8 +344,25 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) } arrow::ArrayVector arrays; std::ranges::for_each(builders, [&arrays](auto& builder) { arrays.push_back(*builder.second->Finish()); }); - auto outTable = arrow::Table::Make(schema, arrays); - allocator.adopt(output, outTable); + // The table is sent by the message backend, followed by the objects it refers to, + // so that they all leave in one multipart and the consumer relays them into the + // same slot. Through the arrow backend, the table would leave on its own first. + auto stream = arrow::io::BufferOutputStream::Create().ValueOrDie(); + auto writer = arrow::ipc::MakeStreamWriter(stream, schema).ValueOrDie(); + if (!writer->WriteRecordBatch(*arrow::RecordBatch::Make(schema, reserveSize, arrays)).ok()) { + LOGP(fatal, "Unable to serialise the table of {}", outBinding); + } + auto buffer = stream->Finish().ValueOrDie(); + auto payload = allocator.makeVector(output); + payload.assign(buffer->data(), buffer->data() + buffer->size()); + allocator.adoptContainer(output, std::move(payload), DataAllocator::CacheStrategy::Never, header::gSerializationMethodArrow); + // Accounted for like the tables of the arrow backend, for the rate limiting. + allocator.findMessageHeader(output)->serialization = header::gSerializationMethodArrow; + arrowContext.updateBytesSent(buffer->size()); + arrowContext.updateMessagesSent(1); + for (auto const& id : referenced) { + allocator.adoptFromCache(output, id, header::gSerializationMethodCCDB); + } } stats.updateStats({(int)ProcessingStatsId::CCDB_CACHE_FETCHED_BYTES, DataProcessingStats::Op::Set, (int64_t)helper->totalFetchedBytes}); diff --git a/Framework/Core/test/test_DataRelayer.cxx b/Framework/Core/test/test_DataRelayer.cxx index 3a5181897892f..6b27cd2f8a2b1 100644 --- a/Framework/Core/test/test_DataRelayer.cxx +++ b/Framework/Core/test/test_DataRelayer.cxx @@ -1105,6 +1105,91 @@ TEST_CASE("DataRelayer") checkContents(); } + // The analysis CCDB fetcher sends a CCDB table followed by the objects it + // refers to, all for the same route. The consumer must get them in the same + // slot, with the table as the first part, which holds when they are relayed + // from one multipart, i.e. without checking for completion in between. + SECTION("CCDBTableAndObjectsCompleteTogether") + { + InputSpec spec0{"bcs", "AOD", "BC"}; + InputSpec spec1{"ccdb", "AOD", "GLOCCDBOBJ", 1}; + + std::vector inputs = { + InputRoute{spec0, 0, "Fake0", 0}, + InputRoute{spec1, 1, "Fake1", 0}, + }; + + std::vector infos{1}; + TimesliceIndex index{1, infos}; + ref.registerService(ServiceRegistryHelpers::handleForService(&index)); + + auto policy = CompletionPolicyHelpers::consumeWhenAll(); + DataRelayer relayer(policy, inputs, index, {registry}, -1); + relayer.setPipelineLength(4); + + auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq"); + auto channelAlloc = o2::pmr::getTransportAllocator(transport.get()); + + auto relayOne = [&](o2::header::DataDescription description, uint32_t subSpec, o2::header::SerializationMethod method, size_t timeslice) { + DataHeader dh; + dh.dataOrigin = "AOD"; + dh.dataDescription = description; + dh.subSpecification = subSpec; + dh.splitPayloadIndex = 0; + dh.splitPayloadParts = 1; + dh.payloadSerializationMethod = method; + std::array msgs; + msgs[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{timeslice, 1}}); + msgs[1] = transport->CreateMessage(8); + DataRelayer::InputInfo info{0, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}}; + relayer.relay(msgs[0]->GetData(), msgs.data(), info, 2); + }; + auto relayCCDB = [&](size_t timeslice) { + relayOne("GLOCCDBOBJ", 1, o2::header::gSerializationMethodArrow, timeslice); + relayOne("GLOCCDBOBJ", 1, o2::header::gSerializationMethodCCDB, timeslice); + relayOne("GLOCCDBOBJ", 1, o2::header::gSerializationMethodCCDB, timeslice); + }; + + // Table and objects in one multipart: one slot, with the table first. + relayOne("BC", 0, o2::header::gSerializationMethodArrow, 0); + relayCCDB(0); + std::vector ready; + relayer.getReadyToProcess(ready); + REQUIRE(ready.size() == 1); + auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); + REQUIRE((result[1] | count_parts{}) == 3); + for (size_t p = 0; p < 3; ++p) { + auto* dh = o2::header::get((result[1] | get_header{p})->GetData()); + REQUIRE(dh->payloadSerializationMethod == (p == 0 ? o2::header::gSerializationMethodArrow : o2::header::gSerializationMethodCCDB)); + } + + // The same, with the CCDB parts arriving before the other input. + relayCCDB(1); + ready.clear(); + relayer.getReadyToProcess(ready); + REQUIRE(ready.empty()); + relayOne("BC", 0, o2::header::gSerializationMethodArrow, 1); + relayer.getReadyToProcess(ready); + REQUIRE(ready.size() == 1); + result = relayer.consumeAllInputsForTimeslice(ready[0].slot); + REQUIRE((result[1] | count_parts{}) == 3); + + // What happened when the table and the objects were sent separately: the + // slot completes on the table alone, and the objects open a new slot which + // can never complete. + relayOne("BC", 0, o2::header::gSerializationMethodArrow, 2); + relayOne("GLOCCDBOBJ", 1, o2::header::gSerializationMethodArrow, 2); + ready.clear(); + relayer.getReadyToProcess(ready); + REQUIRE(ready.size() == 1); + result = relayer.consumeAllInputsForTimeslice(ready[0].slot); + REQUIRE((result[1] | count_parts{}) == 1); + relayOne("GLOCCDBOBJ", 1, o2::header::gSerializationMethodCCDB, 2); + ready.clear(); + relayer.getReadyToProcess(ready); + REQUIRE(ready.empty()); + } + // An expiring input is materialised straight into the slot, so with one // shared buffer per slot it lands *after* whatever the other inputs already // hold -- the cells are then no longer in input order. Check that the data