Skip to content

Commit cd8131b

Browse files
matthiasrichtermatthiasrichter
authored
Introducing configurable channel name for the DPL output proxy (#7304)
The channel name can now be configured via workflow option '--proxy-channel-name' of workflow o2-dpl-output-proxy. The new option defaults to 'downstream' which has been the hardcoded name before. There are probably more cases we need to look into as there are two proxy spec generators in the ExternalFairMQDeviceProxy implementation, a consolidation is planned.
1 parent f440b5c commit cd8131b

2 files changed

Lines changed: 27 additions & 14 deletions

File tree

‎Framework/Core/src/ExternalFairMQDeviceProxy.cxx‎

Lines changed: 22 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
#include "Framework/CallbackService.h"
2222
#include "Framework/ControlService.h"
2323
#include "Framework/SourceInfoHeader.h"
24+
#include "Framework/ConfigParamRegistry.h"
2425
#include "Headers/DataHeader.h"
2526
#include "Headers/Stack.h"
2627

@@ -37,6 +38,7 @@
3738
#include <numeric> // std::accumulate
3839
#include <sstream>
3940
#include <stdexcept>
41+
#include <regex>
4042

4143
namespace o2::framework
4244
{
@@ -438,15 +440,28 @@ DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* name,
438440
spec.name = name;
439441
spec.inputs = inputSpecs;
440442
spec.outputs = {};
441-
spec.algorithm = adaptStateful([inputSpecs](CallbackService& callbacks, RawDeviceService& rds, DeviceSpec const& deviceSpec) {
443+
spec.algorithm = adaptStateful([inputSpecs](CallbackService& callbacks, RawDeviceService& rds, DeviceSpec const& deviceSpec, ConfigParamRegistry const& options) {
444+
// we can retrieve the channel name from the channel configuration string
445+
// FIXME: even if a --channel-config option is specified on the command line, always the default string
446+
// is retrieved from the config registry. The channel name thus needs to be configured in the default
447+
// string AND must match the name in an optional channel config.
448+
std::string channelConfig = options.get<std::string>("channel-config");
449+
std::regex r{R"(name=([^,]*))"};
450+
std::vector<std::string> values{std::sregex_token_iterator{std::begin(channelConfig), std::end(channelConfig), r, 1},
451+
std::sregex_token_iterator{}};
452+
if (values.size() != 1 || values[0].empty()) {
453+
throw std::runtime_error("failed to extract channel name from channel configuration parameter '" + channelConfig + "'");
454+
}
455+
std::string outputChannelName = values[0];
456+
442457
auto* device = rds.device();
443458
// check that the input spec bindings have corresponding output channels
444459
// FairMQDevice calls the custom init before the channels have been configured
445460
// so we do the check before starting in a dedicated callback
446-
auto channelConfigurationChecker = [inputSpecs = std::move(inputSpecs), device]() {
461+
auto channelConfigurationChecker = [inputSpecs = std::move(inputSpecs), device, outputChannelName]() {
447462
LOG(INFO) << "checking channel configuration";
448-
if (device->fChannels.count("downstream") == 0) {
449-
throw std::runtime_error("no corresponding output channel found for input 'downstream'");
463+
if (device->fChannels.count(outputChannelName) == 0) {
464+
throw std::runtime_error("no corresponding output channel found for input '" + outputChannelName + "'");
450465
}
451466
};
452467
callbacks.set(CallbackService::Id::Start, channelConfigurationChecker);
@@ -461,20 +476,17 @@ DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* name,
461476
for (auto const& inputSpec : inputSpecs) {
462477
// this is a prototype, in principle we want to have all spec objects const
463478
// and so only the const object can be retrieved from service registry
464-
ForwardRoute route{0, 1, inputSpec, "downstream"};
479+
ForwardRoute route{0, 1, inputSpec, outputChannelName};
465480
const_cast<DeviceSpec&>(deviceSpec).forwards.emplace_back(route);
466481
}
467482

468-
auto forwardEos = [device, lastDataProcessingHeader](EndOfStreamContext&) {
483+
auto forwardEos = [device, lastDataProcessingHeader, outputChannelName](EndOfStreamContext&) {
469484
// DPL implements an internal end of stream signal, which is propagated through
470485
// all downstream channels if a source is dry, make it available to other external
471486
// devices via a message of type {DPL/EOS/0}
472487
for (auto& channelInfo : device->fChannels) {
473-
// FIXME: in this function the channel name is hardcoded to 'downstream'
474-
// have to check if this simply should be combined with the function below
475-
// supporting multiple outputs
476488
auto& channelName = channelInfo.first;
477-
if (channelName != "downstream") {
489+
if (channelName != outputChannelName) {
478490
continue;
479491
}
480492
DataHeader dh;
@@ -570,9 +582,6 @@ DataProcessorSpec specifyFairMQDeviceMultiOutputProxy(char const* name,
570582
// all downstream channels if a source is dry, make it available to other external
571583
// devices via a message of type {DPL/EOS/0}
572584
for (auto& channelInfo : device->fChannels) {
573-
// FIXME: in this function the channel name is hardcoded to 'downstream'
574-
// have to check if this simply should be combined with the function below
575-
// supporting multiple outputs
576585
auto& channelName = channelInfo.first;
577586
auto checkChannel = [channelNames = std::move(*channelNames)](std::string const& name) -> bool {
578587
for (auto const& n : channelNames) {

‎Framework/Utils/src/dpl-output-proxy.cxx‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,10 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
2525
ConfigParamSpec{
2626
"proxy-name", VariantType::String, "dpl-output-proxy", {"name of the proxy processor, will be the default output channel name as well"}});
2727

28+
workflowOptions.push_back(
29+
ConfigParamSpec{
30+
"proxy-channel-name", VariantType::String, "downstream", {"output channel name of the proxy"}});
31+
2832
workflowOptions.push_back(
2933
ConfigParamSpec{
3034
"dataspec", VariantType::String, "dpl-output-proxy:TST/CLUSTERS;dpl-output-proxy:TST/TRACKS", {"selection string for the data to be proxied"}});
@@ -72,7 +76,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
7276
// vectored options
7377
// use the OutputChannelSpec as a tool to create the default configuration for the out-of-band channel
7478
OutputChannelSpec externalChannelSpec;
75-
externalChannelSpec.name = "downstream";
79+
externalChannelSpec.name = config.options().get<std::string>("proxy-channel-name");
7680
externalChannelSpec.type = ChannelType::Push;
7781
if (config.options().get<std::string>("output-proxy-method") == "bind") {
7882
externalChannelSpec.method = ChannelMethod::Bind;

0 commit comments

Comments
 (0)