Skip to content

Commit dc90288

Browse files
ktfdavidrohr
ktf
authored and
davidrohr
committed
DPL: fix oldest possible timeslice with forwarding
1 parent 046060a commit dc90288

5 files changed

Lines changed: 52 additions & 1 deletion

File tree

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,8 @@ struct ServiceRegistry {
9595
std::vector<ServiceEOSHandle> mPostEOSHandles;
9696
/// Callbacks for services to be executed after every dispatching
9797
std::vector<ServiceDispatchingHandle> mPostDispatchingHandles;
98+
/// Callbacks for services to be executed after every dispatching
99+
std::vector<ServiceForwardingHandle> mPostForwardingHandles;
98100
/// Callbacks for services to be executed before Start
99101
std::vector<ServiceStartHandle> mPreStartHandles;
100102
/// Callbacks for services to be executed on the Stop transition
@@ -131,6 +133,8 @@ struct ServiceRegistry {
131133
/// Invoke callbacks to monitor inputs after dispatching, regardless of them
132134
/// being discarded, consumed or processed.
133135
void postDispatchingCallbacks(ProcessingContext&);
136+
/// Callback invoked after the late forwarding has been done
137+
void postForwardingCallbacks(ProcessingContext&);
134138

135139
/// Invoke callbacks on stop.
136140
void postStopCallbacks();

‎Framework/Core/include/Framework/ServiceSpec.h‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,9 @@ using ServiceMetricHandling = std::function<void(ServiceRegistry&,
9292
/// Callback executed in the child after dispatching happened.
9393
using ServicePostDispatching = std::function<void(ProcessingContext&, void*)>;
9494

95+
/// Callback executed in the child after late forwarding happened.
96+
using ServicePostForwarding = std::function<void(ProcessingContext&, void*)>;
97+
9598
/// Callback invoked when the driver enters the init phase.
9699
using ServiceDriverInit = std::function<void(ServiceRegistry&, boost::program_options::variables_map const&)>;
97100

@@ -153,6 +156,8 @@ struct ServiceSpec {
153156
/// dispatched.
154157
ServicePostDispatching postDispatching = nullptr;
155158

159+
ServicePostForwarding postForwarding = nullptr;
160+
156161
/// Callback invoked on Start
157162
ServiceStartCallback start = nullptr;
158163
/// Callback invoked on Start
@@ -207,6 +212,12 @@ struct ServiceDispatchingHandle {
207212
void* service;
208213
};
209214

215+
struct ServiceForwardingHandle {
216+
ServiceSpec const& spec;
217+
ServicePostForwarding callback;
218+
void* service;
219+
};
220+
210221
struct ServiceStartHandle {
211222
ServiceSpec const& spec;
212223
ServiceStartCallback callback;

‎Framework/Core/src/CommonServices.cxx‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -585,7 +585,7 @@ o2::framework::ServiceSpec CommonServices::decongestionSpec()
585585
}
586586
return ServiceHandle{TypeIdHelpers::uniqueId<DecongestionService>(), decongestion, ServiceKind::Serial};
587587
},
588-
.postProcessing = [](ProcessingContext& ctx, void* service) {
588+
.postForwarding = [](ProcessingContext& ctx, void* service) {
589589
DecongestionService* decongestion = reinterpret_cast<DecongestionService*>(service);
590590
if (decongestion->isFirstInTopology == false) {
591591
LOGP(debug, "We are not the first in the topology, do not update the oldest possible timeslice");
@@ -610,6 +610,18 @@ o2::framework::ServiceSpec CommonServices::decongestionSpec()
610610
oldestPossibleOutput.slot.index == -1 ? "channel" : "slot",
611611
oldestPossibleOutput.slot.index == -1 ? oldestPossibleOutput.channel.value: oldestPossibleOutput.slot.index);
612612
DataProcessingHelpers::broadcastOldestPossibleTimeslice(proxy, oldestPossibleOutput.timeslice.value);
613+
DeviceSpec const& spec = ctx.services().get<DeviceSpec const>();
614+
auto device = ctx.services().get<RawDeviceService>().device();
615+
for (size_t fi = 0; fi < spec.forwards.size(); fi++) {
616+
auto& channel = device->GetChannel(spec.forwards[fi].channel, 0);
617+
// The oldest possible timeslice for a forwarded message
618+
// is conservatively the one of the device doing the forwarding.
619+
if (spec.forwards[fi].channel.rfind("from_", 0) == 0) {
620+
auto oldestTimeslice = timesliceIndex.getOldestPossibleOutput();
621+
DataProcessingHelpers::sendOldestPossibleTimeframe(channel, oldestTimeslice.timeslice.value);
622+
LOGP(debug, "Forwarding to channel {} oldest possible timeslice {}", spec.forwards[fi].channel, oldestTimeslice.timeslice.value);
623+
}
624+
}
613625
decongestion->lastTimeslice = oldestPossibleOutput.timeslice.value; },
614626
.domainInfoUpdated = [](ServiceRegistry& services, size_t oldestPossibleTimeslice, ChannelIndex channel) {
615627
DecongestionService& decongestion = services.get<DecongestionService>();
@@ -632,6 +644,18 @@ o2::framework::ServiceSpec CommonServices::decongestionSpec()
632644
}
633645
LOGP(debug, "Broadcasting possible output {}", oldestPossibleOutput.timeslice.value);
634646
DataProcessingHelpers::broadcastOldestPossibleTimeslice(proxy, oldestPossibleOutput.timeslice.value);
647+
DeviceSpec const& spec = services.get<DeviceSpec const>();
648+
auto device = services.get<RawDeviceService>().device();
649+
for (size_t fi = 0; fi < spec.forwards.size(); fi++) {
650+
auto& channel = device->GetChannel(spec.forwards[fi].channel, 0);
651+
// The oldest possible timeslice for a forwarded message
652+
// is conservatively the one of the device doing the forwarding.
653+
if (spec.forwards[fi].channel.rfind("from_", 0) == 0) {
654+
auto oldestTimeslice = timesliceIndex.getOldestPossibleOutput();
655+
LOGP(info, "Forwarding to channel {} oldest possible timeslice {}", spec.forwards[fi].channel, oldestTimeslice.timeslice.value);
656+
DataProcessingHelpers::sendOldestPossibleTimeframe(channel, oldestTimeslice.timeslice.value);
657+
}
658+
}
635659
decongestion.lastTimeslice = oldestPossibleOutput.timeslice.value; },
636660
.kind = ServiceKind::Serial};
637661
}

‎Framework/Core/src/DataProcessingDevice.cxx‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1889,6 +1889,7 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context,
18891889
LOGP(debug, "Late forwarding");
18901890
forwardInputs(action.slot, record, false, action.op == CompletionPolicy::CompletionOp::Consume);
18911891
}
1892+
context.registry->postForwardingCallbacks(processContext);
18921893
if (action.op == CompletionPolicy::CompletionOp::Consume) {
18931894
#ifdef TRACY_ENABLE
18941895
cleanupRecord(record);

‎Framework/Core/src/ServiceRegistry.cxx‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,9 @@ void ServiceRegistry::bindService(ServiceSpec const& spec, void* service)
125125
if (spec.postDispatching) {
126126
mPostDispatchingHandles.push_back(ServiceDispatchingHandle{spec, spec.postDispatching, service});
127127
}
128+
if (spec.postForwarding) {
129+
mPostForwardingHandles.push_back(ServiceForwardingHandle{spec, spec.postForwarding, service});
130+
}
128131
if (spec.start) {
129132
mPreStartHandles.push_back(ServiceStartHandle{spec, spec.start, service});
130133
}
@@ -194,6 +197,14 @@ void ServiceRegistry::postDispatchingCallbacks(ProcessingContext& processContext
194197
}
195198
}
196199

200+
/// Invoke callbacks to be executed after every data Dispatching
201+
void ServiceRegistry::postForwardingCallbacks(ProcessingContext& processContext)
202+
{
203+
for (auto& forwardingHandle : mPostForwardingHandles) {
204+
forwardingHandle.callback(processContext, forwardingHandle.service);
205+
}
206+
}
207+
197208
/// Callbacks to be called in FairMQDevice::PreRun()
198209
void ServiceRegistry::preStartCallbacks()
199210
{

0 commit comments

Comments
 (0)