Skip to content

Commit 41f5484

Browse files
ktfktf
committed
DPL: deserialised conditions are now cached
1 parent e081d6d commit 41f5484

6 files changed

Lines changed: 106 additions & 14 deletions

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
#include "Framework/ProcessingPolicies.h"
2727
#include "Framework/Tracing.h"
2828
#include "Framework/RunningWorkflowInfo.h"
29+
#include "Framework/ObjectCache.h"
2930

3031
#include <fairmq/FairMQDevice.h>
3132
#include <fairmq/FairMQParts.h>
@@ -86,6 +87,7 @@ struct DataProcessorContext {
8687
AlgorithmSpec::ProcessCallback* statefulProcess = nullptr;
8788
AlgorithmSpec::ProcessCallback* statelessProcess = nullptr;
8889
AlgorithmSpec::ErrorCallback* error = nullptr;
90+
ObjectCache objCache;
8991

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

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

Lines changed: 44 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@
88
// In applying this license CERN does not waive the privileges and immunities
99
// granted to it by virtue of its status as an Intergovernmental Organization
1010
// or submit itself to any jurisdiction.
11-
#ifndef FRAMEWORK_INPUTRECORD_H
12-
#define FRAMEWORK_INPUTRECORD_H
11+
#ifndef O2_FRAMEWORK_INPUTRECORD_H_
12+
#define O2_FRAMEWORK_INPUTRECORD_H_
1313

1414
#include "Framework/DataRef.h"
1515
#include "Framework/DataRefUtils.h"
@@ -18,6 +18,9 @@
1818
#include "Framework/TableConsumer.h"
1919
#include "Framework/Traits.h"
2020
#include "Framework/RuntimeError.h"
21+
#include "Framework/Logger.h"
22+
#include "Framework/ObjectCache.h"
23+
2124
#include "Headers/DataHeader.h"
2225

2326
#include "CommonUtils/BoostSerializer.h"
@@ -34,9 +37,7 @@
3437

3538
#include <fairmq/FwdDecls.h>
3639

37-
namespace o2
38-
{
39-
namespace framework
40+
namespace o2::framework
4041
{
4142

4243
struct InputSpec;
@@ -99,7 +100,8 @@ class InputRecord
99100
using DataHeader = o2::header::DataHeader;
100101

101102
InputRecord(std::vector<InputRoute> const& inputs,
102-
InputSpan& span);
103+
InputSpan& span,
104+
ObjectCache& cache);
103105

104106
/// A deleter type to be used with unique_ptr, which can be marked that
105107
/// it does not own the underlying resource and thus should not delete it.
@@ -384,7 +386,39 @@ class InputRecord
384386
std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<ROOTSerialized<ValueT>>(ref).release());
385387
return result;
386388
} else if (method == o2::header::gSerializationMethodCCDB) {
387-
std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<CCDBSerialized<ValueT>>(ref).release());
389+
// This is to support deserialising objects from CCDB. Contrary to what happens for
390+
// other objects, those objects are most likely long lived, so we
391+
// keep around an instance of the associated object and deserialise it only when
392+
// it's updated.
393+
// FIXME: add ability to apply callbacks to deserialised objects.
394+
auto id = ObjectCache::Id::fromRef(ref);
395+
ConcreteDataMatcher matcher{header->dataOrigin, header->dataDescription, header->subSpecification};
396+
// If the matcher does not have an entry in the cache, deserialise it
397+
// and cache the deserialised object at the given id.
398+
auto path = fmt::format("{}", DataSpecUtils::describe(matcher));
399+
LOGP(info, "{}", path);
400+
auto cacheEntry = mCache.matcherToId.find(path);
401+
if (cacheEntry == mCache.matcherToId.end()) {
402+
mCache.matcherToId.insert(std::make_pair(path, id));
403+
std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<CCDBSerialized<ValueT>>(ref).release(), false);
404+
mCache.idToObject[id] = (void*)result.get();
405+
LOGP(info, "Caching in {} ptr to {} ({})", id.value, path, (void*)result.get());
406+
return result;
407+
}
408+
auto& oldId = cacheEntry->second;
409+
// The id in the cache is the same, let's simply return it.
410+
if (oldId.value == id.value) {
411+
std::unique_ptr<ValueT const, Deleter<ValueT const>> result((ValueT const*)mCache.idToObject[id], false);
412+
LOGP(info, "Returning cached entry {} for {} ({})", id.value, path, (void*)result.get());
413+
return result;
414+
}
415+
// The id in the cache is different. Let's destroy the old cached entry
416+
// and create a new one.
417+
delete reinterpret_cast<ValueT*>(mCache.idToObject[oldId]);
418+
std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<CCDBSerialized<ValueT>>(ref).release(), false);
419+
mCache.idToObject[id] = (void*)result.get();
420+
LOGP(info, "Replacing cached entry {} with {} for {} ({})", oldId.value, id.value, path, (void*)result.get());
421+
oldId.value = id.value;
388422
return result;
389423
} else {
390424
throw runtime_error("Attempt to extract object from message with unsupported serialization type");
@@ -627,11 +661,11 @@ class InputRecord
627661
}
628662

629663
private:
664+
ObjectCache& mCache;
630665
std::vector<InputRoute> const& mInputsSchema;
631666
InputSpan& mSpan;
632667
};
633668

634-
} // namespace framework
635-
} // namespace o2
669+
} // namespace o2::framework
636670

637-
#endif // FRAMEWORK_INPUTREGISTRY_H
671+
#endif // O2_FRAMEWORK_INPUTREGISTRY_H_

‎Framework/Core/include/Framework/ObjectCache.h‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2+
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3+
// All rights not expressly granted are reserved.
4+
//
5+
// This software is distributed under the terms of the GNU General Public
6+
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7+
//
8+
// In applying this license CERN does not waive the privileges and immunities
9+
// granted to it by virtue of its status as an Intergovernmental Organization
10+
// or submit itself to any jurisdiction.
11+
#ifndef O2_FRAMEWORK_OBJECTCACHE_H_
12+
#define O2_FRAMEWORK_OBJECTCACHE_H_
13+
14+
#include "Framework/DataRef.h"
15+
#include <unordered_map>
16+
17+
namespace o2::framework
18+
{
19+
20+
/// A cache for CCDB objects or objects in general
21+
/// which have more than one timeframe of lifetime.
22+
struct ObjectCache {
23+
struct Id {
24+
int64_t value;
25+
static Id fromRef(DataRef& ref)
26+
{
27+
return {reinterpret_cast<int64_t>(ref.payload)};
28+
}
29+
bool operator==(const Id& other) const
30+
{
31+
return value == other.value;
32+
}
33+
34+
struct hash_fn {
35+
std::size_t operator()(const Id& id) const
36+
{
37+
return id.value;
38+
}
39+
};
40+
};
41+
/// A cache for deserialised objects.
42+
/// This keeps a mapping so that we can tell if a given
43+
/// path was already received and it's blob stored in
44+
/// .second.
45+
std::unordered_map<std::string, Id> matcherToId;
46+
/// A map from a CacheId (which is the void* ptr of the previous map).
47+
/// to an actual (type erased) pointer to the deserialised object.
48+
std::unordered_map<Id, void*, Id::hash_fn> idToObject;
49+
};
50+
51+
} // namespace o2::framework
52+
53+
#endif // O2_FRAMEWORK_OBJECTCACHE_H_

‎Framework/Core/src/DataProcessingDevice.cxx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1436,7 +1436,7 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context,
14361436
bool shouldConsume = action.op == CompletionPolicy::CompletionOp::Consume ||
14371437
action.op == CompletionPolicy::CompletionOp::Discard;
14381438
InputSpan span = getInputSpan(action.slot, shouldConsume);
1439-
InputRecord record{context.deviceContext->spec->inputs, span};
1439+
InputRecord record{context.deviceContext->spec->inputs, span, context.objCache};
14401440
ProcessingContext processContext{record, *context.registry, *context.allocator};
14411441
{
14421442
ZoneScopedN("service pre processing");

‎Framework/Core/src/InputRecord.cxx‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
#include "Framework/InputRecord.h"
1212
#include "Framework/InputSpan.h"
1313
#include "Framework/InputSpec.h"
14+
#include "Framework/ObjectCache.h"
1415
#include <fairmq/FairMQMessage.h>
1516
#include <cassert>
1617

@@ -32,9 +33,11 @@ namespace o2::framework
3233
{
3334

3435
InputRecord::InputRecord(std::vector<InputRoute> const& inputsSchema,
35-
InputSpan& span)
36+
InputSpan& span,
37+
ObjectCache& cache)
3638
: mInputsSchema{inputsSchema},
37-
mSpan{span}
39+
mSpan{span},
40+
mCache{cache}
3841
{
3942
}
4043

‎Framework/TestWorkflows/src/test_CCDBFetcher.cxx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
4040
for (size_t pi = 0; pi < condition->size(); pi++) {
4141
LOGP(info, "Phase at {} for timestamp {} is {}", pi, condition->timestamp(pi), condition->LHCphase(pi));
4242
}
43-
control.readyToQuit(QuitRequest::All);
43+
// control.readyToQuit(QuitRequest::All);
4444
})},
4545
Options{
4646
{"test-option", VariantType::String, "test", {"A test option"}}},

0 commit comments

Comments
 (0)