Skip to content

Commit 71404a6

Browse files
ktfktf
committed
DPL: invoke callback on deserialization
1 parent 41f5484 commit 71404a6

5 files changed

Lines changed: 70 additions & 21 deletions

File tree

‎Framework/Core/include/Framework/CallbackService.h‎

Lines changed: 17 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ struct DataHeader;
2626
namespace o2::framework
2727
{
2828

29+
struct ConcreteDataMatcher;
2930
class EndOfStreamContext;
3031

3132
// A service that data processors can register callback functions invoked by the
@@ -69,7 +70,8 @@ class CallbackService
6970
/// to a timeslice with the wanted quantities.
7071
NewTimeslice,
7172
/// Invoked before the processing callback
72-
PreProcessing
73+
PreProcessing,
74+
CCDBDeserialised
7375
};
7476

7577
using StartCallback = std::function<void()>;
@@ -82,19 +84,21 @@ class CallbackService
8284
using RegionInfoCallback = std::function<void(FairMQRegionInfo const&)>;
8385
using NewTimesliceCallback = std::function<void(o2::header::DataHeader&, DataProcessingHeader&)>;
8486
using PreProcessingCallback = std::function<void(ServiceRegistry&, int)>;
87+
using CCDBDeserializedCallback = std::function<void(ConcreteDataMatcher&, void*)>;
8588

86-
using Callbacks = CallbackRegistry<Id, //
87-
RegistryPair<Id, Id::Start, StartCallback>, //
88-
RegistryPair<Id, Id::Stop, StopCallback>, //
89-
RegistryPair<Id, Id::Reset, ResetCallback>, //
90-
RegistryPair<Id, Id::Idle, IdleCallback>, //
91-
RegistryPair<Id, Id::ClockTick, ClockTickCallback>, //
92-
RegistryPair<Id, Id::DataConsumed, DataConsumedCallback>, //
93-
RegistryPair<Id, Id::EndOfStream, EndOfStreamCallback>, //
94-
RegistryPair<Id, Id::RegionInfoCallback, RegionInfoCallback>, //
95-
RegistryPair<Id, Id::NewTimeslice, NewTimesliceCallback>, //
96-
RegistryPair<Id, Id::PreProcessing, PreProcessingCallback> //
97-
>; //
89+
using Callbacks = CallbackRegistry<Id, //
90+
RegistryPair<Id, Id::Start, StartCallback>, //
91+
RegistryPair<Id, Id::Stop, StopCallback>, //
92+
RegistryPair<Id, Id::Reset, ResetCallback>, //
93+
RegistryPair<Id, Id::Idle, IdleCallback>, //
94+
RegistryPair<Id, Id::ClockTick, ClockTickCallback>, //
95+
RegistryPair<Id, Id::DataConsumed, DataConsumedCallback>, //
96+
RegistryPair<Id, Id::EndOfStream, EndOfStreamCallback>, //
97+
RegistryPair<Id, Id::RegionInfoCallback, RegionInfoCallback>, //
98+
RegistryPair<Id, Id::NewTimeslice, NewTimesliceCallback>, //
99+
RegistryPair<Id, Id::PreProcessing, PreProcessingCallback>, //
100+
RegistryPair<Id, Id::CCDBDeserialised, CCDBDeserializedCallback> //
101+
>; //
98102

99103
// set callback for specified processing step
100104
template <typename U>

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

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#include "Framework/RuntimeError.h"
2121
#include "Framework/Logger.h"
2222
#include "Framework/ObjectCache.h"
23+
#include "Framework/CallbackService.h"
2324

2425
#include "Headers/DataHeader.h"
2526

@@ -42,6 +43,7 @@ namespace o2::framework
4243

4344
struct InputSpec;
4445
struct InputSpan;
46+
struct CallbackService;
4547

4648
/// @class InputRecord
4749
/// @brief The input API of the Data Processing Layer
@@ -101,7 +103,8 @@ class InputRecord
101103

102104
InputRecord(std::vector<InputRoute> const& inputs,
103105
InputSpan& span,
104-
ObjectCache& cache);
106+
ObjectCache& cache,
107+
CallbackService& callbacks);
105108

106109
/// A deleter type to be used with unique_ptr, which can be marked that
107110
/// it does not own the underlying resource and thus should not delete it.
@@ -401,8 +404,10 @@ class InputRecord
401404
if (cacheEntry == mCache.matcherToId.end()) {
402405
mCache.matcherToId.insert(std::make_pair(path, id));
403406
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());
407+
void* obj = (void*)result.get();
408+
mCallbacks(CallbackService::Id::CCDBDeserialised, matcher, obj);
409+
mCache.idToObject[id] = obj;
410+
LOGP(info, "Caching in {} ptr to {} ({})", id.value, path, obj);
406411
return result;
407412
}
408413
auto& oldId = cacheEntry->second;
@@ -416,8 +421,10 @@ class InputRecord
416421
// and create a new one.
417422
delete reinterpret_cast<ValueT*>(mCache.idToObject[oldId]);
418423
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());
424+
void* obj = (void*)result.get();
425+
mCallbacks(CallbackService::Id::CCDBDeserialised, matcher, obj);
426+
mCache.idToObject[id] = obj;
427+
LOGP(info, "Replacing cached entry {} with {} for {} ({})", oldId.value, id.value, path, obj);
421428
oldId.value = id.value;
422429
return result;
423430
} else {
@@ -664,6 +671,7 @@ class InputRecord
664671
ObjectCache& mCache;
665672
std::vector<InputRoute> const& mInputsSchema;
666673
InputSpan& mSpan;
674+
CallbackService& mCallbacks;
667675
};
668676

669677
} // namespace o2::framework

‎Framework/Core/include/Framework/Task.h‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,24 @@ class has_endOfStream
3838
enum { value = sizeof(test<T>(nullptr)) == sizeof(char) };
3939
};
4040

41+
/// Check if the class task has EndOfStream
42+
template <typename T>
43+
class has_finaliseCCDB
44+
{
45+
typedef char one;
46+
struct two {
47+
char x[2];
48+
};
49+
50+
template <typename C>
51+
static one test(decltype(&C::finaliseCCDB));
52+
template <typename C>
53+
static two test(...);
54+
55+
public:
56+
enum { value = sizeof(test<T>(nullptr)) == sizeof(char) };
57+
};
58+
4159
/// Check if the class task has Stop
4260
template <typename T>
4361
class has_stop
@@ -79,6 +97,13 @@ class Task
7997
/// This is invoked whenever we have an EndOfStream event
8098
virtual void endOfStream(EndOfStreamContext& context) {}
8199

100+
/// This is invoked whenever a new CCDB object associated to
101+
/// a given ConcreteDataMatcher is deserialised
102+
virtual void finaliseCCDB(ConcreteDataMatcher& matcher, void* obj)
103+
{
104+
LOGP(error, "CCDB deserialization invoked");
105+
}
106+
82107
/// This is invoked on stop
83108
virtual void stop() {}
84109
};
@@ -96,6 +121,12 @@ AlgorithmSpec adaptFromTask(Args&&... args)
96121
task->endOfStream(eosContext);
97122
});
98123
}
124+
if constexpr (has_finaliseCCDB<T>::value) {
125+
auto& callbacks = ic.services().get<CallbackService>();
126+
callbacks.set(CallbackService::Id::CCDBDeserialised, [task](ConcreteDataMatcher& matcher, void* obj) {
127+
task->finaliseCCDB(matcher, obj);
128+
});
129+
}
99130
if constexpr (has_stop<T>::value) {
100131
auto& callbacks = ic.services().get<CallbackService>();
101132
callbacks.set(CallbackService::Id::Stop, [task]() {

‎Framework/Core/src/DataProcessingDevice.cxx‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1436,7 +1436,10 @@ 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, context.objCache};
1439+
InputRecord record{context.deviceContext->spec->inputs,
1440+
span,
1441+
context.objCache,
1442+
context.registry->get<CallbackService>()};
14401443
ProcessingContext processContext{record, *context.registry, *context.allocator};
14411444
{
14421445
ZoneScopedN("service pre processing");

‎Framework/Core/src/InputRecord.cxx‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
#include "Framework/InputSpan.h"
1313
#include "Framework/InputSpec.h"
1414
#include "Framework/ObjectCache.h"
15+
#include "Framework/CallbackService.h"
1516
#include <fairmq/FairMQMessage.h>
1617
#include <cassert>
1718

@@ -34,10 +35,12 @@ namespace o2::framework
3435

3536
InputRecord::InputRecord(std::vector<InputRoute> const& inputsSchema,
3637
InputSpan& span,
37-
ObjectCache& cache)
38+
ObjectCache& cache,
39+
CallbackService& callbacks)
3840
: mInputsSchema{inputsSchema},
3941
mSpan{span},
40-
mCache{cache}
42+
mCache{cache},
43+
mCallbacks{callbacks}
4144
{
4245
}
4346

0 commit comments

Comments
 (0)