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

Refactor the examples after move from FairRoot

Browse Source
This commit is contained in:
Alexey Rybalchenko
2018-04-26 23:06:01 +02:00
parent 71b2981526
commit 7aefa0eeb8
148 changed files with 1582 additions and 2027 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
examples/dds/CMakeLists.txt Normal file
View File

@@ -0,0 +1,60 @@
################################################################################
# Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
# #
# This software is distributed under the terms of the #
# GNU Lesser General Public Licence (LGPL) version 3, #
# copied verbatim in the file "LICENSE" #
################################################################################
add_library(ExampleDDSLib STATIC
"Sampler.cxx"
"Sampler.h"
"Processor.cxx"
"Processor.h"
"Sink.cxx"
"Sink.h"
)
target_link_libraries(ExampleDDSLib PUBLIC FairMQ)
add_executable(fairmq-ex-dds-sampler runSampler.cxx)
target_link_libraries(fairmq-ex-dds-sampler PRIVATE ExampleDDSLib)
add_executable(fairmq-ex-dds-processor runProcessor.cxx)
target_link_libraries(fairmq-ex-dds-processor PRIVATE ExampleDDSLib)
add_executable(fairmq-ex-dds-sink runSink.cxx)
target_link_libraries(fairmq-ex-dds-sink PRIVATE ExampleDDSLib)
add_custom_target(ExampleDDS DEPENDS fairmq-ex-dds-sampler fairmq-ex-dds-processor fairmq-ex-dds-sink)
install(
TARGETS
fairmq-ex-dds-sampler
fairmq-ex-dds-processor
fairmq-ex-dds-sink
LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR}
RUNTIME DESTINATION ${PROJECT_INSTALL_BINDIR}
)
# configure run script with different executable paths for build and for install directories
set(EX_BIN_DIR ${CMAKE_CURRENT_BINARY_DIR})
set(DDS_PLUGIN_LIB_DIR ${CMAKE_BINARY_DIR}/fairmq/plugins/DDS)
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/ex-dds-topology.xml ${CMAKE_CURRENT_BINARY_DIR}/ex-dds-topology.xml @ONLY)
set(EX_BIN_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_BINDIR})
set(DDS_PLUGIN_LIB_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_LIBDIR})
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/ex-dds-topology.xml ${CMAKE_CURRENT_BINARY_DIR}/ex-dds-topology.xml_install @ONLY)
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/ex-dds-hosts.cfg ${CMAKE_CURRENT_BINARY_DIR}/ex-dds-hosts.cfg COPYONLY)
install(
FILES ${CMAKE_CURRENT_BINARY_DIR}/ex-dds-topology.xml_install
DESTINATION ${PROJECT_INSTALL_BINDIR}
RENAME ex-dds-topology.xml
)
install(
FILES ${CMAKE_CURRENT_BINARY_DIR}/ex-dds-hosts.cfg
DESTINATION ${PROJECT_INSTALL_BINDIR}
)

52
examples/dds/Processor.cxx Normal file
View File

@@ -0,0 +1,52 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "Processor.h"
#include "FairMQLogger.h"
using namespace std;
namespace example_dds
{
Processor::Processor()
{
OnData("data1", &Processor::HandleData);
}
bool Processor::HandleData(FairMQMessagePtr& msg, int /*index*/)
{
LOG(info) << "Received data, processing...";
// Modify the received string
string* text = new std::string(static_cast<char*>(msg->GetData()), msg->GetSize());
*text += " (modified by " + fId + ")";
// create message object with a pointer to the data buffer,
// its size,
// custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer
FairMQMessagePtr msg2(NewMessage(const_cast<char*>(text->c_str()),
text->length(),
[](void* /*data*/, void* object) { delete static_cast<string*>(object); },
text));
// Send out the output message
if (Send(msg2, "data2") < 0)
{
return false;
}
return true;
}
Processor::~Processor()
{
}
} // namespace example_dds

29
examples/dds/Processor.h Normal file
View File

@@ -0,0 +1,29 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQEXAMPLEDDSPROCESSOR_H
#define FAIRMQEXAMPLEDDSPROCESSOR_H
#include "FairMQDevice.h"
namespace example_dds
{
class Processor : public FairMQDevice
{
public:
Processor();
virtual ~Processor();
protected:
bool HandleData(FairMQMessagePtr&, int);
};
}
#endif /* FAIRMQEXAMPLEDDSPROCESSOR_H */

107
examples/dds/README.md Normal file
View File

@@ -0,0 +1,107 @@
Example 3: DDS
===============
This example demonstrates usage of the Dynamic Deployment System ([DDS](http://dds.gsi.de/)) to dynamically deploy and configure a topology of devices. The topology is similar to those of Example 2, but now it can be easily distributed on different computing nodes without the need for manual socket reconfiguration of the devices.
To make use of DDS functionality the example executables need to find the DDS plugin libraries that are compiled with FairRoot when FairRoot find DDS installed. DDS location is given to CMake as follows:
```bash
cmake -DDDS_PATH="/path/to/dds/install/dir/" ..
```
The description below outlines the minimal steps needed to run the example with DDS. For general DDS help please refer to DDS documentation on [DDS Website](http://dds.gsi.de/).
##### 1. The device handles the socket addresses and ports configuration via DDS Plugin.
It is sufficient to provide the `-S "<@FAIRROOT_INSTALL_DIR@/lib" -P dds` (`<` prepends the following path to the default plugin search paths; put in the path which points to the library dir of your FairRoot installation) command line arguments to let the devices be configured dynamically. No code changes in the device are necessary. See the XML topology file for example of using the command line arguments.
##### 2a. Write DDS hosts file that contains a list of worker nodes to run the topology on (When deploying using the SSH plug-in).
We run this example on the local machine for simplicity. The file below defines three workers, sampler, processor and sink, with a total of 12 DDS agents (thus able to accept 12 tasks). The parameters for each worker node are:
- user-chosen worker ID (must be unique)
- a host name with or without a login, in a form: login@host.fqdn (password-less SSH access to these hosts must be possible)
- additional SSH params (can be empty)
- a remote working directory (most exist on the worker nodes)
- number of DDS Agents for this worker
```bash
@bash_begin@
#source setup.sh
@bash_end@
sampler, username@localhost, , /path/to/dds-work-dir/, 1
processor, username@localhost, , /path/to/dds-work-dir/, 10
sink, username@localhost, , /path/to/dds-work-dir/, 1
```
##### 2b. Skip (When deploying using the localhost plug-in).
If you want to deploy on a single host DDS 1.6+ provides a localhost rms plug-in. You do not need a hosts file in that case.
##### 3. Write DDS topology file that describes which tasks (processes) to run and their topology and configuration.
Take a look at `ex3-dds-topology.xml`. It consists of a definition part (properties, tasks, collections and more) and execution part (main). In our example Sampler, Processor and Sink tasks are defines, containing their executables and exchanged properties. The `<main>` of the topology uses the defined tasks. Besides one Sampler and one Sink task, a group containing Processor task is defined. The group has a multiplicity of 10, meaninig 10 Processors will be executed. Each of the Processors will receive the properties with Sampler and Sink addresses.
The configuration of the channel connection addresses is done by the DDS plugin via the channel names. The task property names must correspond to the channel names (data1, data2), with binding channels writing the properties and connecting channel reading the properties (see the example XML and JSON files).
If `eth0` network interface (default for binding) is not available on your system, specify another one in the topology file for each task. For example: `--network-interface lo0`.
If you chose step 2b earlier, then modify the provided `ex3-dds-topology.xml` in the top that the following lines read as following:
```xml
<declrequirement id="SamplerWorker" type="wnname" value=".*"/>
<declrequirement id="ProcessorWorker" type="wnname" value=".*"/>
<declrequirement id="SinkWorker" type="wnname" value=".*"/>
```
Note that the attributes `value` contain a different value.
##### 4. Start DDS server.
The DDS server is started with:
```bash
dds-server start -s
```
##### 5. Submit DDS Agents (configured in the hosts file).
Agents are submitted with:
```bash
dds-submit --rms ssh --config ex3-dds-hosts.cfg
```
The `--rms` option defines a destination resource management system. The `--config` specifies an SSH plug-in resource definition file.
If you chose step 2b earlier, run the following command instead:
```bash
dds-submit --rms localhost -n 12
```
##### 6. Activate the topology.
```bash
dds-topology --activate ex3-dds-topology.xml
```
##### 7. Run
After activation, agents will execute the defined tasks on the worker nodes. Output of the tasks will be stored in the directory that was specified in the hosts file.
##### 8. (optional) Use example command UI to check state of the devices
A simple utility (fairmq-dds-command-ui) is included with FairRoot to send commands to devices and receive replies from them. The utility uses the DDS intercom library to send "check-state" string to all devices, to which they reply with their ID and state they are in. The utility also allows requesting state changes from devices. To let the device listen to the commands from the utility, start the device with `-S "<@FAIRROOT_INSTALL_DIR@/lib" -P dds` cmd option (see example XML topology).
To see it in action, start the fairmq-dds-command-ui while the topology is running.
##### 9. Stop DDS server/topology.
The execution of tasks can be stopped with:
```bash
dds-topology --stop
```
Or by stopping the DDS server:
```bash
dds-server stop
```
For general DDS documentation please refer to [DDS Website](http://dds.gsi.de/).

53
examples/dds/Sampler.cxx Normal file
View File

@@ -0,0 +1,53 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* Sampler.cpp
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#include <thread> // this_thread::sleep_for
#include <chrono>
#include "Sampler.h"
using namespace std;
namespace example_dds
{
Sampler::Sampler()
{
}
bool Sampler::ConditionalRun()
{
std::this_thread::sleep_for(std::chrono::seconds(1));
// NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place).
// Should only be used for small data because of the cost of an additional copy
FairMQMessagePtr msg(NewSimpleMessage("Data"));
LOG(info) << "Sending \"Data\"";
// in case of error or transfer interruption, return false to go to IDLE state
// successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
if (Send(msg, "data1") < 0)
{
return false;
}
return true;
}
Sampler::~Sampler()
{
}
}

35
examples/dds/Sampler.h Normal file
View File

@@ -0,0 +1,35 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* Sampler.h
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#ifndef FAIRMQEXAMPLEDDSSAMPLER_H
#define FAIRMQEXAMPLEDDSSAMPLER_H
#include "FairMQDevice.h"
namespace example_dds
{
class Sampler : public FairMQDevice
{
public:
Sampler();
virtual ~Sampler();
protected:
virtual bool ConditionalRun();
};
} // namespace example_dds
#endif /* FAIRMQEXAMPLEDDSSAMPLER_H */

41
examples/dds/Sink.cxx Normal file
View File

@@ -0,0 +1,41 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* Sink.cxx
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#include "Sink.h"
using namespace std;
namespace example_dds
{
Sink::Sink()
{
// register a handler for data arriving on "data2" channel
OnData("data2", &Sink::HandleData);
}
// handler is called whenever a message arrives on "data2", with a reference to the message and a sub-channel index (here 0)
bool Sink::HandleData(FairMQMessagePtr& msg, int /*index*/)
{
LOG(info) << "Received: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
// return true if want to be called again (otherwise go to IDLE state)
return true;
}
Sink::~Sink()
{
}
} // namespace example_dds

35
examples/dds/Sink.h Normal file
View File

@@ -0,0 +1,35 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* Sink.h
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#ifndef FAIRMQEXAMPLEDDSSINK_H
#define FAIRMQEXAMPLEDDSSINK_H
#include "FairMQDevice.h"
namespace example_dds
{
class Sink : public FairMQDevice
{
public:
Sink();
virtual ~Sink();
protected:
bool HandleData(FairMQMessagePtr&, int);
};
} // namespace example_dds
#endif /* FAIRMQEXAMPLEDDSSINK_H */

7
examples/dds/ex-dds-hosts.cfg Normal file
View File

@@ -0,0 +1,7 @@
@bash_begin@
# source setup.sh
@bash_end@
sampler, username@localhost, , /path/to/dds-work-dir/, 1
processor, username@localhost, , /path/to/dds-work-dir/, 10
sink, username@localhost, , /path/to/dds-work-dir/, 1

49
examples/dds/ex-dds-topology.xml Normal file
View File

@@ -0,0 +1,49 @@
<topology id="ExampleDDS">
<property id="data1" />
<property id="data2" />
<declrequirement id="SamplerWorker" type="wnname" value="sampler"/>
<declrequirement id="ProcessorWorker" type="wnname" value="processor"/>
<declrequirement id="SinkWorker" type="wnname" value="sink"/>
<decltask id="Sampler">
<exe reachable="true">@EX_BIN_DIR@/fairmq-ex-dds-sampler --id sampler --color false --channel-config name=data1,type=push,method=bind -S "&lt;@DDS_PLUGIN_LIB_DIR@/" -P dds</exe>
<requirements>
<id>SamplerWorker</id>
</requirements>
<properties>
<id access="write">data1</id>
</properties>
</decltask>
<decltask id="Processor">
<exe reachable="true">@EX_BIN_DIR@/fairmq-ex-dds-processor --id processor_%taskIndex% --config-key processor --color false --channel-config name=data1,type=pull,method=connect name=data2,type=push,method=connect -S "&lt;@DDS_PLUGIN_LIB_DIR@/" -P dds</exe>
<requirements>
<id>ProcessorWorker</id>
</requirements>
<properties>
<id access="read">data1</id>
<id access="read">data2</id>
</properties>
</decltask>
<decltask id="Sink">
<exe reachable="true">@EX_BIN_DIR@/fairmq-ex-dds-sink --id sink --color false --channel-config name=data2,type=pull,method=bind -S "&lt;@DDS_PLUGIN_LIB_DIR@/" -P dds</exe>
<requirements>
<id>SinkWorker</id>
</requirements>
<properties>
<id access="write">data2</id>
</properties>
</decltask>
<main id="main">
<task>Sampler</task>
<task>Sink</task>
<group id="ProcessorGroup" n="10">
<task>Processor</task>
</group>
</main>
</topology>

21
examples/dds/runProcessor.cxx Normal file
View File

@@ -0,0 +1,21 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "runFairMQDevice.h"
#include "Processor.h"
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& /*options*/)
{
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
{
return new example_dds::Processor();
}

21
examples/dds/runSampler.cxx Normal file
View File

@@ -0,0 +1,21 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "runFairMQDevice.h"
#include "Sampler.h"
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& /*options*/)
{
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
{
return new example_dds::Sampler();
}

21
examples/dds/runSink.cxx Normal file
View File

@@ -0,0 +1,21 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "runFairMQDevice.h"
#include "Sink.h"
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& /*options*/)
{
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
{
return new example_dds::Sink();
}