Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 00:31:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

DDS Plugin: Simplify subchannel bookkeeping

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-03-11 14:37:08 +01:00 committed by Dennis Klein
parent b05782af16
commit 539b088ade
2 changed files with 10 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
fairmq/plugins/DDS/DDS.cxx
View File

@ -165,9 +165,7 @@ auto DDS::FillChannelContainers() -> void
} else if (GetProperty<string>(methodKey) == "connect") {
fConnectingChans.insert(make_pair(c.first, DDSConfig()));
LOG(debug) << "preparing to connect: " << c.first << " with " << c.second << " sub-channels.";
for (int i = 0; i < c.second; ++i) {
fConnectingChans.at(c.first).fSubChannelAddresses.push_back(string());
}
fConnectingChans.at(c.first).fNumSubChannels = c.second;
} else {
LOG(error) << "Cannot update address configuration. Channel method (bind/connect) not specified.";
return;
@ -273,19 +271,13 @@ auto DDS::SubscribeForConnectingChannels() -> void
fConnectingChans.at(channelName).fDDSValues.insert({senderTaskID, val.c_str()});
}
// update channels and remove them from unfinished container
for (auto mi = fConnectingChans.begin(); mi != fConnectingChans.end(); /* no increment */) {
if (mi->second.fSubChannelAddresses.size() == mi->second.fDDSValues.size()) {
// when multiple subChannels are used, their order on every device should be the same, irregardless of arrival order from DDS.
sort(mi->second.fSubChannelAddresses.begin(), mi->second.fSubChannelAddresses.end());
auto it3 = mi->second.fDDSValues.begin();
for (unsigned int i = 0; i < mi->second.fSubChannelAddresses.size(); ++i) {
SetProperty<string>(string{"chans." + mi->first + "." + to_string(i) + ".address"}, it3->second);
++it3;
for (const auto& mi : fConnectingChans) {
if (mi.second.fNumSubChannels == mi.second.fDDSValues.size()) {
int i = 0;
for (const auto& e : mi.second.fDDSValues) {
SetProperty<string>(string{"chans." + mi.first + "." + to_string(i) + ".address"}, e.second);
++i;
}
fConnectingChans.erase(mi++);
} else {
++mi;
}
}
} catch (const exception& e) {

5
fairmq/plugins/DDS/DDS.h
View File

@ -27,6 +27,7 @@
#include <string>
#include <atomic>
#include <thread>
#include <map>
#include <unordered_map>
#include <vector>
@ -40,9 +41,9 @@ namespace plugins
struct DDSConfig
{
// container of sub channel addresses
std::vector<std::string> fSubChannelAddresses;
unsigned int fNumSubChannels;
// dds values for the channel
std::unordered_map<uint64_t, std::string> fDDSValues;
std::map<uint64_t, std::string> fDDSValues;
};
struct DDSSubscription