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

DDS plugin: Automatically set session and device id if not provided

Browse Source 
Resolves #187
This commit is contained in:
Dennis Klein 2019-07-28 19:20:11 +02:00 committed by Dennis Klein
parent 07fe02a0a0
commit 1bb77bf47b
6 changed files with 42 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
examples/dds/ex-dds-topology-infinite.xml
View File

@ -8,8 +8,8 @@
<declrequirement name="SinkWorker" type="wnname" value="sink"/>
<decltask name="Sampler">
<exe>fairmq-ex-dds-sampler --color false --channel-config name=data1,type=push,method=bind --rate 100 -P dds</exe>
<env reachable="false">fairmq-ex-dds-env.sh</env>
<exe>fairmq-ex-dds-sampler --id sampler --color false --channel-config name=data1,type=push,method=bind --rate 100 -P dds</exe>
<requirements>
<name>SamplerWorker</name>
</requirements>
@ -19,8 +19,8 @@
</decltask>
<decltask name="Processor">
<exe>fairmq-ex-dds-processor --color false --channel-config name=data1,type=pull,method=connect name=data2,type=push,method=connect -P dds</exe>
<env reachable="false">fairmq-ex-dds-env.sh</env>
<exe>fairmq-ex-dds-processor --id processor_%taskIndex% --color false --channel-config name=data1,type=pull,method=connect name=data2,type=push,method=connect -P dds</exe>
<requirements>
<id>ProcessorWorker</id>
</requirements>
@ -31,8 +31,8 @@
</decltask>
<decltask name="Sink">
<exe>fairmq-ex-dds-sink --color false --channel-config name=data2,type=pull,method=bind -P dds</exe>
<env reachable="false">fairmq-ex-dds-env.sh</env>
<exe>fairmq-ex-dds-sink --id sink --color false --channel-config name=data2,type=pull,method=bind -P dds</exe>
<requirements>
<name>SinkWorker</name>
</requirements>

6
examples/dds/ex-dds-topology.xml
View File

@ -8,8 +8,8 @@
<declrequirement name="SinkWorker" type="wnname" value="sink"/>
<decltask name="Sampler">
<exe>fairmq-ex-dds-sampler --color false --channel-config name=data1,type=push,method=bind -P dds --iterations 10</exe>
<env reachable="false">fairmq-ex-dds-env.sh</env>
<exe>fairmq-ex-dds-sampler --id sampler --color false --channel-config name=data1,type=push,method=bind -P dds --iterations 10</exe>
<requirements>
<name>SamplerWorker</name>
</requirements>
@ -19,8 +19,8 @@
</decltask>
<decltask name="Processor">
<exe>fairmq-ex-dds-processor --color false --channel-config name=data1,type=pull,method=connect name=data2,type=push,method=connect -P dds</exe>
<env reachable="false">fairmq-ex-dds-env.sh</env>
<exe>fairmq-ex-dds-processor --id processor_%taskIndex% --color false --channel-config name=data1,type=pull,method=connect name=data2,type=push,method=connect -P dds</exe>
<requirements>
<id>ProcessorWorker</id>
</requirements>
@ -31,8 +31,8 @@
</decltask>
<decltask name="Sink">
<exe>fairmq-ex-dds-sink --color false --channel-config name=data2,type=pull,method=bind -P dds --iterations 10</exe>
<env reachable="false">fairmq-ex-dds-env.sh</env>
<exe>fairmq-ex-dds-sink --id sink --color false --channel-config name=data2,type=pull,method=bind -P dds --iterations 10</exe>
<requirements>
<name>SinkWorker</name>
</requirements>

5
fairmq/DeviceRunner.cxx
View File

@ -63,8 +63,6 @@ bool DeviceRunner::HandleGeneralOptions(const fair::mq::ProgOptions& config, boo
<< " / __/ / /_/ / / / _ / / / / /_/ / " << FAIRMQ_REPO_URL << endl
<< " /_/ \\__,_/_/_/ /_/ /_/ \\___\\_\\ " << FAIRMQ_LICENSE << " © " << FAIRMQ_COPYRIGHT << endl;
}
config.PrintOptions();
}
return true;
@ -169,6 +167,9 @@ auto DeviceRunner::Run() -> int
// Instantiate and run plugins
fPluginManager.InstantiatePlugins();
// Log IDLE configuration
fConfig.PrintOptions();
// Run the device
fDevice->RunStateMachine();

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

@ -55,18 +55,18 @@ DDS::DDS(const string& name,
{
try {
TakeDeviceControl();
fControllerThread = thread(&DDS::HandleControl, this);
fHeartbeatThread = thread(&DDS::HeartbeatSender, this);
} catch (PluginServices::DeviceControlError& e) {
LOG(debug) << e.what();
} catch (exception& e) {
LOG(error) << "Error in plugin initialization: " << e.what();
}
}
auto DDS::HandleControl() -> void
{
try {
fHeartbeatThread = thread(&DDS::HeartbeatSender, this);
std::string deviceId(GetProperty<std::string>("id"));
if (deviceId.empty()) {
SetProperty<std::string>("id", dds::env_prop<dds::task_path>());
}
std::string sessionId(GetProperty<std::string>("session"));
if (sessionId == "default") {
SetProperty<std::string>("session", dds::env_prop<dds::dds_session_id>());
}
auto control = GetProperty<string>("control");
bool staticMode(false);
if (control == "static") {
@ -121,15 +121,26 @@ auto DDS::HandleControl() -> void
});
if (staticMode) {
TransitionDeviceStateTo(DeviceState::Running);
// wait until stop signal
unique_lock<mutex> lock(fStopMutex);
while (!fDeviceTerminationRequested) {
fStopCondition.wait_for(lock, chrono::seconds(1));
}
LOG(debug) << "Stopping DDS control plugin";
fControllerThread = thread(&DDS::StaticControl, this);
}
} catch (PluginServices::DeviceControlError& e) {
LOG(debug) << e.what();
} catch (exception& e) {
LOG(error) << "Error in plugin initialization: " << e.what();
}
}
auto DDS::StaticControl() -> void
{
try {
TransitionDeviceStateTo(DeviceState::Running);
// wait until stop signal
unique_lock<mutex> lock(fStopMutex);
while (!fDeviceTerminationRequested) {
fStopCondition.wait_for(lock, chrono::seconds(1));
}
LOG(debug) << "Stopping DDS plugin static controller";
} catch (DeviceErrorState&) {
ReleaseDeviceControl();
} catch (exception& e) {

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

@ -128,7 +128,7 @@ class DDS : public Plugin
~DDS();
private:
auto HandleControl() -> void;
auto StaticControl() -> void;
auto FillChannelContainers() -> void;
auto SubscribeForConnectingChannels() -> void;

4
test/sdk/test_topo.xml
View File

@ -6,7 +6,7 @@
<declrequirement name="SinkWorker" type="wnname" value="sink"/>
<decltask name="Sampler">
<exe reachable="true">fairmq-bsampler --id sampler --color false --channel-config name=data,type=push,method=bind -P dds --msg-rate 10</exe>
<exe reachable="true">fairmq-bsampler --color false --channel-config name=data,type=push,method=bind -P dds --msg-rate 10</exe>
<requirements>
<name>SamplerWorker</name>
</requirements>
@ -16,7 +16,7 @@
</decltask>
<decltask name="Sink">
<exe reachable="true">fairmq-sink --id sink_%taskIndex% --color false --channel-config name=data,type=pull,method=connect -P dds</exe>
<exe reachable="true">fairmq-sink --color false --channel-config name=data,type=pull,method=connect -P dds</exe>
<requirements>
<name>SinkWorker</name>
</requirements>