Skip to content

Commit 46aef9b

Browse files
ktfktf
committed
DPL: introduce ObjectCache as a service
Avoid exposing ObjectCache and CallbackService to the world via InputRecord.
1 parent 2be7771 commit 46aef9b

12 files changed

Lines changed: 54 additions & 55 deletions

‎Framework/Core/include/Framework/CommonServices.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ struct CommonServices {
6767
static ServiceSpec tracingSpec();
6868
static ServiceSpec threadPool(int numWorkers);
6969
static ServiceSpec dataProcessingStats();
70+
static ServiceSpec objectCache();
7071
static ServiceSpec timingInfoSpec();
7172

7273
static std::vector<ServiceSpec> defaultServices(int numWorkers = 0);

‎Framework/Core/include/Framework/DataProcessingDevice.h‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,6 @@ struct DataProcessorContext {
8787
AlgorithmSpec::ProcessCallback* statefulProcess = nullptr;
8888
AlgorithmSpec::ProcessCallback* statelessProcess = nullptr;
8989
AlgorithmSpec::ErrorCallback* error = nullptr;
90-
ObjectCache objCache;
9190

9291
/// Wether or not the associated DataProcessor can forward things early
9392
bool canForwardEarly = true;

‎Framework/Core/include/Framework/InputRecord.h‎

Lines changed: 13 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -103,8 +103,7 @@ class InputRecord
103103

104104
InputRecord(std::vector<InputRoute> const& inputs,
105105
InputSpan& span,
106-
ObjectCache& cache,
107-
CallbackService& callbacks);
106+
ServiceRegistry&);
108107

109108
/// A deleter type to be used with unique_ptr, which can be marked that
110109
/// it does not own the underlying resource and thus should not delete it.
@@ -400,30 +399,32 @@ class InputRecord
400399
// and cache the deserialised object at the given id.
401400
auto path = fmt::format("{}", DataSpecUtils::describe(matcher));
402401
LOGP(info, "{}", path);
403-
auto cacheEntry = mCache.matcherToId.find(path);
404-
if (cacheEntry == mCache.matcherToId.end()) {
405-
mCache.matcherToId.insert(std::make_pair(path, id));
402+
auto& cache = mRegistry.get<ObjectCache>();
403+
auto& callbacks = mRegistry.get<CallbackService>();
404+
auto cacheEntry = cache.matcherToId.find(path);
405+
if (cacheEntry == cache.matcherToId.end()) {
406+
cache.matcherToId.insert(std::make_pair(path, id));
406407
std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<CCDBSerialized<ValueT>>(ref).release(), false);
407408
void* obj = (void*)result.get();
408-
mCallbacks(CallbackService::Id::CCDBDeserialised, (ConcreteDataMatcher&)matcher, (void*)obj);
409-
mCache.idToObject[id] = obj;
409+
callbacks(CallbackService::Id::CCDBDeserialised, (ConcreteDataMatcher&)matcher, (void*)obj);
410+
cache.idToObject[id] = obj;
410411
LOGP(info, "Caching in {} ptr to {} ({})", id.value, path, obj);
411412
return result;
412413
}
413414
auto& oldId = cacheEntry->second;
414415
// The id in the cache is the same, let's simply return it.
415416
if (oldId.value == id.value) {
416-
std::unique_ptr<ValueT const, Deleter<ValueT const>> result((ValueT const*)mCache.idToObject[id], false);
417+
std::unique_ptr<ValueT const, Deleter<ValueT const>> result((ValueT const*)cache.idToObject[id], false);
417418
LOGP(info, "Returning cached entry {} for {} ({})", id.value, path, (void*)result.get());
418419
return result;
419420
}
420421
// The id in the cache is different. Let's destroy the old cached entry
421422
// and create a new one.
422-
delete reinterpret_cast<ValueT*>(mCache.idToObject[oldId]);
423+
delete reinterpret_cast<ValueT*>(cache.idToObject[oldId]);
423424
std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<CCDBSerialized<ValueT>>(ref).release(), false);
424425
void* obj = (void*)result.get();
425-
mCallbacks(CallbackService::Id::CCDBDeserialised, (ConcreteDataMatcher&)matcher, (void*)obj);
426-
mCache.idToObject[id] = obj;
426+
callbacks(CallbackService::Id::CCDBDeserialised, (ConcreteDataMatcher&)matcher, (void*)obj);
427+
cache.idToObject[id] = obj;
427428
LOGP(info, "Replacing cached entry {} with {} for {} ({})", oldId.value, id.value, path, obj);
428429
oldId.value = id.value;
429430
return result;
@@ -438,8 +439,6 @@ class InputRecord
438439
} else {
439440
// non-messageable objects for which serialization method can not be derived by type,
440441
// the operation depends on the transmitted serialization method
441-
using DataHeader = o2::header::DataHeader;
442-
443442
auto header = DataRefUtils::getHeader<header::DataHeader*>(ref);
444443
auto method = header->payloadSerializationMethod;
445444
if (method == o2::header::gSerializationMethodNone) {
@@ -668,10 +667,9 @@ class InputRecord
668667
}
669668

670669
private:
671-
ObjectCache& mCache;
670+
ServiceRegistry& mRegistry;
672671
std::vector<InputRoute> const& mInputsSchema;
673672
InputSpan& mSpan;
674-
CallbackService& mCallbacks;
675673
};
676674

677675
} // namespace o2::framework

‎Framework/Core/src/CommonServices.cxx‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -629,6 +629,18 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats()
629629
.kind = ServiceKind::Serial};
630630
}
631631

632+
o2::framework::ServiceSpec CommonServices::objectCache()
633+
{
634+
return ServiceSpec{
635+
.name = "object-cache",
636+
.init = [](ServiceRegistry&, DeviceState&, fair::mq::ProgOptions&) -> ServiceHandle {
637+
auto* cache = new ObjectCache();
638+
return ServiceHandle{TypeIdHelpers::uniqueId<ObjectCache>(), cache};
639+
},
640+
.configure = noConfiguration(),
641+
.kind = ServiceKind::Serial};
642+
}
643+
632644
std::vector<ServiceSpec> CommonServices::defaultServices(int numThreads)
633645
{
634646
std::vector<ServiceSpec> specs{
@@ -647,6 +659,7 @@ std::vector<ServiceSpec> CommonServices::defaultServices(int numThreads)
647659
dataRelayer(),
648660
dataSender(),
649661
dataProcessingStats(),
662+
objectCache(),
650663
CommonMessageBackends::fairMQBackendSpec(),
651664
ArrowSupport::arrowBackendSpec(),
652665
CommonMessageBackends::stringBackendSpec(),

‎Framework/Core/src/DataProcessingDevice.cxx‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1454,8 +1454,7 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context,
14541454
InputSpan span = getInputSpan(action.slot, shouldConsume);
14551455
InputRecord record{context.deviceContext->spec->inputs,
14561456
span,
1457-
context.objCache,
1458-
context.registry->get<CallbackService>()};
1457+
*context.registry};
14591458
ProcessingContext processContext{record, *context.registry, *context.allocator};
14601459
{
14611460
ZoneScopedN("service pre processing");

‎Framework/Core/src/InputRecord.cxx‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -35,12 +35,10 @@ namespace o2::framework
3535

3636
InputRecord::InputRecord(std::vector<InputRoute> const& inputsSchema,
3737
InputSpan& span,
38-
ObjectCache& cache,
39-
CallbackService& callbacks)
40-
: mInputsSchema{inputsSchema},
41-
mSpan{span},
42-
mCache{cache},
43-
mCallbacks{callbacks}
38+
ServiceRegistry& registry)
39+
: mRegistry{registry},
40+
mInputsSchema{inputsSchema},
41+
mSpan{span}
4442
{
4543
}
4644

‎Framework/Core/test/benchmark_InputRecord.cxx‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -48,9 +48,8 @@ static void BM_InputRecordGenericGetters(benchmark::State& state)
4848
// First of all we test if an empty registry behaves as expected, raising a
4949
// bunch of exceptions.
5050
InputSpan span{[](size_t) { return DataRef{nullptr, nullptr, nullptr}; }, 0};
51-
CallbackService callbacks;
52-
ObjectCache cache;
53-
InputRecord emptyRecord(schema, span, cache, callbacks);
51+
ServiceRegistry registry;
52+
InputRecord emptyRecord(schema, span, registry);
5453

5554
std::vector<void*> inputs;
5655

@@ -84,17 +83,17 @@ static void BM_InputRecordGenericGetters(benchmark::State& state)
8483
createMessage(dh2, 2);
8584
createEmpty();
8685
InputSpan span2{[&inputs](size_t i) { return DataRef{nullptr, static_cast<char const*>(inputs[2 * i]), static_cast<char const*>(inputs[2 * i + 1])}; }, inputs.size() / 2};
87-
InputRecord record{schema, span2, cache, callbacks};
86+
InputRecord record{schema, span2, registry};
8887

8988
for (auto _ : state) {
9089
// Checking we can get the whole ref by name
91-
auto ref00 = record.get("x");
92-
auto ref10 = record.get("y");
93-
auto ref20 = record.get("z");
90+
[[maybe_unused]] auto ref00 = record.get("x");
91+
[[maybe_unused]] auto ref10 = record.get("y");
92+
[[maybe_unused]] auto ref20 = record.get("z");
9493

9594
// Or we can get it positionally
96-
auto ref01 = record.getByPos(0);
97-
auto ref11 = record.getByPos(1);
95+
[[maybe_unused]] auto ref01 = record.getByPos(0);
96+
[[maybe_unused]] auto ref11 = record.getByPos(1);
9897

9998
record.isValid("x");
10099
record.isValid("y");

‎Framework/Core/test/test_InputRecord.cxx‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -51,9 +51,8 @@ BOOST_AUTO_TEST_CASE(TestInputRecord)
5151
InputSpan span{
5252
[](size_t) { return DataRef{nullptr, nullptr, nullptr}; },
5353
0};
54-
CallbackService callbacks;
55-
ObjectCache cache;
56-
InputRecord emptyRecord(schema, span, cache, callbacks);
54+
ServiceRegistry registry;
55+
InputRecord emptyRecord(schema, span, registry);
5756

5857
BOOST_CHECK_EXCEPTION(emptyRecord.get("x"), RuntimeErrorRef, any_exception);
5958
BOOST_CHECK_EXCEPTION(emptyRecord.get("y"), RuntimeErrorRef, any_exception);
@@ -95,7 +94,7 @@ BOOST_AUTO_TEST_CASE(TestInputRecord)
9594
createMessage(dh2, 2);
9695
createEmpty();
9796
InputSpan span2{[&inputs](size_t i) { return DataRef{nullptr, static_cast<char const*>(inputs[2 * i]), static_cast<char const*>(inputs[2 * i + 1])}; }, inputs.size() / 2};
98-
InputRecord record{schema, span2, cache, callbacks};
97+
InputRecord record{schema, span2, registry};
9998

10099
// Checking we can get the whole ref by name
101100
BOOST_CHECK_NO_THROW(record.get("x"));

‎Framework/Core/test/test_InputRecordWalker.cxx‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ struct DataSet {
3939
using TaggedSet = std::pair<o2::header::DataOrigin, MessageSet>;
4040
using Messages = std::vector<TaggedSet>;
4141
using CheckType = std::vector<std::string>;
42-
DataSet(std::vector<InputRoute>&& s, Messages&& m, CheckType&& v, ObjectCache& cache, CallbackService& callbacks)
42+
DataSet(std::vector<InputRoute>&& s, Messages&& m, CheckType&& v, ServiceRegistry& registry)
4343
: schema{std::move(s)}, messages{std::move(m)}, span{[this](size_t i, size_t part) {
4444
BOOST_REQUIRE(i < this->messages.size());
4545
BOOST_REQUIRE(part < this->messages[i].second.size() / 2);
@@ -48,7 +48,7 @@ struct DataSet {
4848
return DataRef{nullptr, header, payload};
4949
},
5050
[this](size_t i) { return i < this->messages.size() ? messages[i].second.size() / 2 : 0; }, this->messages.size()},
51-
record{schema, span, cache, callbacks},
51+
record{schema, span, registry},
5252
values{std::move(v)}
5353
{
5454
BOOST_REQUIRE(messages.size() == schema.size());
@@ -63,8 +63,7 @@ struct DataSet {
6363

6464
DataSet createData()
6565
{
66-
static CallbackService callbacks;
67-
static ObjectCache cache;
66+
static ServiceRegistry registry;
6867
// Create the routes we want for the InputRecord
6968
std::vector<InputSpec> inputspecs = {
7069
InputSpec{"tpc", "TPC", "SOMEDATA", 0, Lifetime::Timeframe},
@@ -140,7 +139,7 @@ DataSet createData()
140139
createMessage(dh3);
141140
createMessage(dh4);
142141

143-
return {std::move(schema), std::move(messages), std::move(checkValues), cache, callbacks};
142+
return {std::move(schema), std::move(messages), std::move(checkValues), registry};
144143
}
145144

146145
BOOST_AUTO_TEST_CASE(test_DPLRawParser)

‎Framework/Utils/test/RawPageTestData.cxx‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -86,9 +86,8 @@ DataSet createData(std::vector<InputSpec> const& inputspecs, std::vector<DataHea
8686
}
8787
}
8888

89-
static ObjectCache cache;
90-
static CallbackService callbacks;
91-
return {std::move(schema), std::move(messages), std::move(checkValues), cache, callbacks};
89+
static ServiceRegistry registry;
90+
return {std::move(schema), std::move(messages), std::move(checkValues), registry};
9291
}
9392

9493
} // namespace test

0 commit comments

Comments
 (0)