Skip to content

Commit f5ff838

Browse files
ktfktf
committed
DPL: Fix salt and use globalSalt
1 parent 37b06fc commit f5ff838

4 files changed

Lines changed: 74 additions & 32 deletions

File tree

‎Framework/Core/include/Framework/ServiceRegistry.h‎

Lines changed: 40 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -112,10 +112,45 @@ struct ServiceRegistry {
112112
/// The mask to use to calculate the initial slot id.
113113
constexpr static uint32_t MAX_SERVICES_MASK = MAX_SERVICES - 1;
114114

115-
static Salt threadSalt() {
116-
auto tid = std::this_thread::get_id();
117-
std::hash<std::thread::id> hasher;
118-
return Salt{Context{.streamId = (short)hasher(tid)}};
115+
/// A salt which is global to the whole device.
116+
/// This can be used to query services which are not
117+
/// bound to a specific stream or data processor, e.g.
118+
/// the services to send metrics to the driver or
119+
/// to send messages to the control.
120+
static Salt globalDeviceSalt()
121+
{
122+
return GLOBAL_CONTEXT_SALT;
123+
}
124+
125+
/// A salt which is global to a given stream
126+
/// but which multiple dataprocessors can share.
127+
static Salt globalStreamSalt(short streamId)
128+
{
129+
// FIXME: old behaviour for now
130+
// return {streamId, 0};
131+
return GLOBAL_CONTEXT_SALT;
132+
}
133+
134+
/// A salt which is global to a specific data processor.
135+
/// This can be used to query properties which are
136+
/// not bonded to a specific stream, e.g. the
137+
/// name of the data processor, its inputs and outputs,
138+
/// it's algorithm.
139+
static Salt dataProcessorSalt(short dataProcessorId)
140+
{
141+
// FIXME: old behaviour for now
142+
// return {0, dataProcessorId};
143+
return GLOBAL_CONTEXT_SALT;
144+
}
145+
146+
/// A salt which is specific to a given stream.
147+
/// This can be used to query properties which are of the stream
148+
/// itself, e.g. the currently processed time frame by a given stream.
149+
static Salt streamSalt(short streamId, short dataProcessorId)
150+
{
151+
// FIXME: old behaviour for now
152+
// return {streamId, dataProcessorId};
153+
return GLOBAL_CONTEXT_SALT;
119154
}
120155

121156
constexpr InstanceId instanceFromTypeSalt(ServiceTypeHash type, Salt salt) const
@@ -211,7 +246,7 @@ struct ServiceRegistry {
211246
/// thread safe.
212247
/// @a salt is used to create the service in the proper context
213248
/// FIXME: for now we create everything in the global context
214-
void declareService(ServiceSpec const& spec, DeviceState& state, fair::mq::ProgOptions& options, ServiceRegistry::Salt salt = ServiceRegistry::threadSalt());
249+
void declareService(ServiceSpec const& spec, DeviceState& state, fair::mq::ProgOptions& options, ServiceRegistry::Salt salt = ServiceRegistry::globalDeviceSalt());
215250

216251
/// Bind the callbacks of a service spec to a given service.
217252
void bindService(ServiceSpec const& spec, void* service);

‎Framework/Core/include/Framework/ServiceRegistryRef.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,9 +36,9 @@ class ServiceRegistryRef
3636
// cannot be accessed if the streamId is <= 0 and complain accordingly.
3737
// The dataProcessorId will be used to distinguish between different
3838
// data processors when
39-
ServiceRegistryRef(ServiceRegistry& registry)
39+
ServiceRegistryRef(ServiceRegistry& registry, ServiceRegistry::Salt salt = ServiceRegistry::globalDeviceSalt())
4040
: mRegistry(registry),
41-
mSalt(ServiceRegistry::threadSalt())
41+
mSalt(salt)
4242
{
4343
}
4444

‎Framework/Core/src/DataProcessingDevice.cxx‎

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -113,38 +113,44 @@ void on_communication_requested(uv_async_t* s)
113113
}
114114

115115
DataProcessingDevice::DataProcessingDevice(RunningDeviceRef ref, ServiceRegistry& registry, ProcessingPolicies& policies)
116-
: mSpec{registry.get<RunningWorkflowInfo const>(ServiceRegistry::threadSalt()).devices[ref.index]},
117-
mState{registry.get<DeviceState>(ServiceRegistry::threadSalt())},
116+
: mSpec{registry.get<RunningWorkflowInfo const>(ServiceRegistry::globalDeviceSalt()).devices[ref.index]},
117+
mState{registry.get<DeviceState>(ServiceRegistry::globalDeviceSalt())},
118118
mInit{mSpec.algorithm.onInit},
119119
mStatefulProcess{nullptr},
120120
mStatelessProcess{mSpec.algorithm.onProcess},
121121
mError{mSpec.algorithm.onError},
122122
mConfigRegistry{nullptr},
123123
mServiceRegistry{registry},
124124
mProcessingPolicies{policies},
125-
mQuotaEvaluator{registry.get<ComputingQuotaEvaluator>(ServiceRegistry::threadSalt())}
125+
mQuotaEvaluator{registry.get<ComputingQuotaEvaluator>(ServiceRegistry::globalDeviceSalt())}
126126
{
127127

128128
/// FIXME: move erro handling to a service?
129129
if (mError != nullptr) {
130130
mErrorHandling = [&errorCallback = mError,
131131
&serviceRegistry = mServiceRegistry](RuntimeErrorRef e, InputRecord& record) {
132132
ZoneScopedN("Error handling");
133+
/// FIXME: we should pass the salt in, so that the message
134+
/// can access information which were stored in the stream.
135+
ServiceRegistryRef ref{serviceRegistry, ServiceRegistry::globalDeviceSalt()};
133136
auto& err = error_from_ref(e);
134137
LOGP(error, "Exception caught: {} ", err.what);
135138
demangled_backtrace_symbols(err.backtrace, err.maxBacktrace, STDERR_FILENO);
136-
serviceRegistry.get<DataProcessingStats>(ServiceRegistry::threadSalt()).exceptionCount++;
137-
ErrorContext errorContext{record, serviceRegistry, e};
139+
ref.get<DataProcessingStats>().exceptionCount++;
140+
ErrorContext errorContext{record, ref, e};
138141
errorCallback(errorContext);
139142
};
140143
} else {
141144
mErrorHandling = [&errorPolicy = mProcessingPolicies.error,
142145
&serviceRegistry = mServiceRegistry](RuntimeErrorRef e, InputRecord& record) {
143146
ZoneScopedN("Error handling");
144147
auto& err = error_from_ref(e);
148+
/// FIXME: we should pass the salt in, so that the message
149+
/// can access information which were stored in the stream.
145150
LOGP(error, "Exception caught: {} ", err.what);
151+
ServiceRegistryRef ref{serviceRegistry, ServiceRegistry::globalDeviceSalt()};
146152
demangled_backtrace_symbols(err.backtrace, err.maxBacktrace, STDERR_FILENO);
147-
serviceRegistry.get<DataProcessingStats>(ServiceRegistry::threadSalt()).exceptionCount++;
153+
ref.get<DataProcessingStats>().exceptionCount++;
148154
switch (errorPolicy) {
149155
case TerminationPolicy::QUIT:
150156
throw e;
@@ -155,9 +161,10 @@ DataProcessingDevice::DataProcessingDevice(RunningDeviceRef ref, ServiceRegistry
155161
}
156162

157163
std::function<void(const fair::mq::State)> stateWatcher = [this, &registry = mServiceRegistry](const fair::mq::State state) -> void {
158-
auto& deviceState = registry.get<DeviceState>(ServiceRegistry::threadSalt());
159-
auto& control = registry.get<ControlService>(ServiceRegistry::threadSalt());
160-
auto& callbacks = registry.get<CallbackService>(ServiceRegistry::threadSalt());
164+
auto ref = ServiceRegistryRef{registry, ServiceRegistry::globalDeviceSalt()};
165+
auto& deviceState = ref.get<DeviceState>();
166+
auto& control = ref.get<ControlService>();
167+
auto& callbacks = ref.get<CallbackService>();
161168
control.notifyDeviceState(fair::mq::GetStateName(state));
162169
callbacks(CallbackService::Id::DeviceStateChanged, registry, state);
163170

@@ -356,7 +363,7 @@ void DataProcessingDevice::Init()
356363
str = entry.second.get_value<std::string>();
357364
}
358365
std::string configString = fmt::format("[CONFIG] {}={} 1 {}", entry.first, str, configStore->provenance(entry.first.c_str())).c_str();
359-
mServiceRegistry.get<DriverClient>(ServiceRegistry::threadSalt()).tell(configString.c_str());
366+
mServiceRegistry.get<DriverClient>(ServiceRegistry::globalDeviceSalt()).tell(configString.c_str());
360367
}
361368

362369
mConfigRegistry = std::make_unique<ConfigParamRegistry>(std::move(configStore));
@@ -386,7 +393,7 @@ void DataProcessingDevice::Init()
386393
// Invoke the callback policy for this device.
387394
if (mSpec.callbacksPolicy.policy != nullptr) {
388395
InitContext initContext{*mConfigRegistry, mServiceRegistry};
389-
mSpec.callbacksPolicy.policy(mServiceRegistry.get<CallbackService>(ServiceRegistry::threadSalt()), initContext);
396+
mSpec.callbacksPolicy.policy(mServiceRegistry.get<CallbackService>(ServiceRegistry::globalDeviceSalt()), initContext);
390397
}
391398
}
392399

@@ -864,7 +871,7 @@ void DataProcessingDevice::InitTask()
864871
// more a ServiceRegistry::globalDataProcessorSalt(N) where
865872
// N is the number of the multiplexed data processor.
866873
// We will get there.
867-
this->fillContext(mServiceRegistry.get<DataProcessorContext>(ServiceRegistry::threadSalt()), deviceContext);
874+
this->fillContext(mServiceRegistry.get<DataProcessorContext>(ServiceRegistry::globalDeviceSalt()), deviceContext);
868875

869876
/// We now run an event loop also in InitTask. This is needed to:
870877
/// * Make sure region registration callbacks are invoked

‎Framework/Core/test/test_Services.cxx‎

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -57,14 +57,14 @@ BOOST_AUTO_TEST_CASE(TestServiceRegistry)
5757
ref.registerService(ServiceRegistryHelpers::handleForService<InterfaceA>(&serviceA));
5858
ref.registerService(ServiceRegistryHelpers::handleForService<InterfaceB>(&serviceB));
5959
ref.registerService(ServiceRegistryHelpers::handleForService<InterfaceC const>(&serviceC));
60-
BOOST_CHECK(registry.get<InterfaceA>(ServiceRegistry::threadSalt()).method() == true);
61-
BOOST_CHECK(registry.get<InterfaceB>(ServiceRegistry::threadSalt()).method() == false);
62-
BOOST_CHECK(registry.get<InterfaceC const>(ServiceRegistry::threadSalt()).method() == false);
63-
BOOST_CHECK(registry.active<InterfaceA>(ServiceRegistry::threadSalt()) == true);
64-
BOOST_CHECK(registry.active<InterfaceB>(ServiceRegistry::threadSalt()) == true);
65-
BOOST_CHECK(registry.active<InterfaceC>(ServiceRegistry::threadSalt()) == false);
66-
BOOST_CHECK_THROW(registry.get<InterfaceA const>(ServiceRegistry::threadSalt()), RuntimeErrorRef);
67-
BOOST_CHECK_THROW(registry.get<InterfaceC>(ServiceRegistry::threadSalt()), RuntimeErrorRef);
60+
BOOST_CHECK(registry.get<InterfaceA>(ServiceRegistry::globalDeviceSalt()).method() == true);
61+
BOOST_CHECK(registry.get<InterfaceB>(ServiceRegistry::globalDeviceSalt()).method() == false);
62+
BOOST_CHECK(registry.get<InterfaceC const>(ServiceRegistry::globalDeviceSalt()).method() == false);
63+
BOOST_CHECK(registry.active<InterfaceA>(ServiceRegistry::globalDeviceSalt()) == true);
64+
BOOST_CHECK(registry.active<InterfaceB>(ServiceRegistry::globalDeviceSalt()) == true);
65+
BOOST_CHECK(registry.active<InterfaceC>(ServiceRegistry::globalDeviceSalt()) == false);
66+
BOOST_CHECK_THROW(registry.get<InterfaceA const>(ServiceRegistry::globalDeviceSalt()), RuntimeErrorRef);
67+
BOOST_CHECK_THROW(registry.get<InterfaceC>(ServiceRegistry::globalDeviceSalt()), RuntimeErrorRef);
6868
}
6969

7070
BOOST_AUTO_TEST_CASE(TestCallbackService)
@@ -78,13 +78,13 @@ BOOST_AUTO_TEST_CASE(TestCallbackService)
7878
// the callback simply sets the captured variable to indicated that it was called
7979
bool cbCalled = false;
8080
auto cb = [&]() { cbCalled = true; };
81-
registry.get<CallbackService>(ServiceRegistry::threadSalt()).set(CallbackService::Id::Stop, cb);
81+
registry.get<CallbackService>(ServiceRegistry::globalDeviceSalt()).set(CallbackService::Id::Stop, cb);
8282

8383
// check to set with the wrong type
84-
BOOST_CHECK_THROW(registry.get<CallbackService>(ServiceRegistry::threadSalt()).set(CallbackService::Id::Stop, [](int) {}), RuntimeErrorRef);
84+
BOOST_CHECK_THROW(registry.get<CallbackService>(ServiceRegistry::globalDeviceSalt()).set(CallbackService::Id::Stop, [](int) {}), RuntimeErrorRef);
8585

8686
// execute and check
87-
registry.get<CallbackService>(ServiceRegistry::threadSalt())(CallbackService::Id::Stop);
87+
registry.get<CallbackService>(ServiceRegistry::globalDeviceSalt())(CallbackService::Id::Stop);
8888
BOOST_CHECK(cbCalled);
8989
}
9090

@@ -191,8 +191,8 @@ BOOST_AUTO_TEST_CASE(TestServiceDeclaration)
191191
options.SetProperty("configuration", "command-line");
192192

193193
registry.declareService(CommonServices::callbacksSpec(), state, options);
194-
BOOST_CHECK(registry.active<CallbackService>(ServiceRegistry::threadSalt()) == true);
195-
BOOST_CHECK(registry.active<DummyService>(ServiceRegistry::threadSalt()) == false);
194+
BOOST_CHECK(registry.active<CallbackService>(ServiceRegistry::globalDeviceSalt()) == true);
195+
BOOST_CHECK(registry.active<DummyService>(ServiceRegistry::globalDeviceSalt()) == false);
196196
}
197197

198198
BOOST_AUTO_TEST_CASE(TestServiceOverride)

0 commit comments

Comments
 (0)