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

Adding multiple transports support & other fixes:

Browse Source 
- Avoid polling when only one input channel is used.
 - Send only handles for shared memory transport.
 - Avoid waiting in the rate logger thread when nothing to log.
 - Hide warnings from generated files
 - Fix #483
This commit is contained in:
Alexey Rybalchenko 2017-01-13 15:53:25 +01:00 committed by Mohammad Al-Turany
parent 22abc3671b
commit 8defd77b63
20 changed files with 509 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/MQ/1-sampler-sink/CMakeLists.txt
View File

@ -13,8 +13,6 @@ configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/1-sampler-sink/startMQEx1.sh.in
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/zeromq
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options

2
examples/MQ/2-sampler-processor-sink/CMakeLists.txt
View File

@ -13,8 +13,6 @@ configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/2-sampler-processor-sink/startMQE
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/zeromq
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options

2
examples/MQ/3-dds/CMakeLists.txt
View File

@ -15,8 +15,6 @@ configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/3-dds/ex3-dds.json
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/zeromq
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options

2
examples/MQ/4-copypush/CMakeLists.txt
View File

@ -13,8 +13,6 @@ configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/4-copypush/startMQEx4.sh.in
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/zeromq
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options

2
examples/MQ/5-req-rep/CMakeLists.txt
View File

@ -13,8 +13,6 @@ configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/5-req-rep/startMQEx5.sh.in
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/zeromq
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options

2
examples/MQ/6-multiple-channels/CMakeLists.txt
View File

@ -13,8 +13,6 @@ configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/6-multiple-channels/startMQEx6.sh
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/zeromq
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options

2
examples/MQ/8-multipart/CMakeLists.txt
View File

@ -13,8 +13,6 @@ configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/8-multipart/startMQEx8.sh.in
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/zeromq
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options

82
examples/MQ/multiple-transports/CMakeLists.txt Normal file
View File

@ -0,0 +1,82 @@
################################################################################
# Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
# #
# This software is distributed under the terms of the #
# GNU Lesser General Public Licence version 3 (LGPL) version 3, #
# copied verbatim in the file "LICENSE" #
################################################################################
configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports/ex-multiple-transports.json
${CMAKE_BINARY_DIR}/bin/config/ex-multiple-transports.json)
configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports/startMTEx.sh.in
${CMAKE_BINARY_DIR}/bin/examples/MQ/multiple-transports/startMTEx.sh)
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/devices
${CMAKE_SOURCE_DIR}/fairmq/tools
${CMAKE_SOURCE_DIR}/fairmq/options
${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports
${CMAKE_CURRENT_BINARY_DIR}
)
Set(SYSTEM_INCLUDE_DIRECTORIES
${Boost_INCLUDE_DIR}
${ZMQ_INCLUDE_DIR}
)
Include_Directories(${INCLUDE_DIRECTORIES})
Include_Directories(SYSTEM ${SYSTEM_INCLUDE_DIRECTORIES})
Set(LINK_DIRECTORIES
${Boost_LIBRARY_DIRS}
)
Link_Directories(${LINK_DIRECTORIES})
Set(SRCS
"FairMQExampleMTSampler1.cxx"
"FairMQExampleMTSampler2.cxx"
"FairMQExampleMTSink.cxx"
)
Set(DEPENDENCIES
${DEPENDENCIES}
FairMQ
)
Set(LIBRARY_NAME FairMQExampleMT)
GENERATE_LIBRARY()
Set(Exe_Names
ex-mt-sampler1
ex-mt-sampler2
ex-mt-sink
)
Set(Exe_Source
runExampleMTSampler1.cxx
runExampleMTSampler2.cxx
runExampleMTSink.cxx
)
list(LENGTH Exe_Names _length)
math(EXPR _length ${_length}-1)
set(BIN_DESTINATION share/fairbase/examples/MQ/multiple-transports/bin)
set(EXECUTABLE_OUTPUT_PATH "${EXECUTABLE_OUTPUT_PATH}/examples/MQ/multiple-transports")
ForEach(_file RANGE 0 ${_length})
list(GET Exe_Names ${_file} _name)
list(GET Exe_Source ${_file} _src)
Set(EXE_NAME ${_name})
Set(SRCS ${_src})
Set(DEPENDENCIES FairMQExampleMT)
GENERATE_EXECUTABLE()
EndForEach(_file RANGE 0 ${_length})
Install(
FILES ex-multiple-transports.json
DESTINATION share/fairbase/examples/MQ/multiple-transports/config/
)

68
examples/MQ/multiple-transports/FairMQExampleMTSampler1.cxx Normal file
View File

@ -0,0 +1,68 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "FairMQExampleMTSampler1.h"
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
using namespace std;
FairMQExampleMTSampler1::FairMQExampleMTSampler1()
: fAckListener()
{
}
void FairMQExampleMTSampler1::InitTask()
{
}
void FairMQExampleMTSampler1::PreRun()
{
fAckListener = thread(&FairMQExampleMTSampler1::ListenForAcks, this);
}
bool FairMQExampleMTSampler1::ConditionalRun()
{
// Creates a message using the transport of channel data1
FairMQMessagePtr msg(NewMessageFor("data1", 0, 1000000));
// 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;
}
void FairMQExampleMTSampler1::PostRun()
{
fAckListener.join();
}
void FairMQExampleMTSampler1::ListenForAcks()
{
uint64_t numAcks = 0;
while (CheckCurrentState(RUNNING))
{
FairMQMessagePtr ack(NewMessageFor("ack", 0));
if (Receive(ack, "ack") < 0)
{
break;
}
++numAcks;
}
LOG(INFO) << "Acknowledged " << numAcks << " messages";
}
FairMQExampleMTSampler1::~FairMQExampleMTSampler1()
{
}

33
examples/MQ/multiple-transports/FairMQExampleMTSampler1.h Normal file
View File

@ -0,0 +1,33 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQEXAMPLEMTSAMPLER1_H_
#define FAIRMQEXAMPLEMTSAMPLER1_H_
#include <string>
#include <thread>
#include "FairMQDevice.h"
class FairMQExampleMTSampler1 : public FairMQDevice
{
public:
FairMQExampleMTSampler1();
virtual ~FairMQExampleMTSampler1();
protected:
virtual void InitTask();
virtual void PreRun();
virtual bool ConditionalRun();
virtual void PostRun();
void ListenForAcks();
std::thread fAckListener;
};
#endif /* FAIRMQEXAMPLEMTSAMPLER1_H_ */

35
examples/MQ/multiple-transports/FairMQExampleMTSampler2.cxx 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 version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "FairMQExampleMTSampler2.h"
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
using namespace std;
FairMQExampleMTSampler2::FairMQExampleMTSampler2()
{
}
bool FairMQExampleMTSampler2::ConditionalRun()
{
FairMQMessagePtr msg(NewMessage(1000));
// 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, "data2") < 0)
{
return false;
}
return true;
}
FairMQExampleMTSampler2::~FairMQExampleMTSampler2()
{
}

27
examples/MQ/multiple-transports/FairMQExampleMTSampler2.h Normal file
View File

@ -0,0 +1,27 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQEXAMPLEMTSAMPLER2_H_
#define FAIRMQEXAMPLEMTSAMPLER2_H_
#include <string>
#include <thread>
#include "FairMQDevice.h"
class FairMQExampleMTSampler2 : public FairMQDevice
{
public:
FairMQExampleMTSampler2();
virtual ~FairMQExampleMTSampler2();
protected:
virtual bool ConditionalRun();
};
#endif /* FAIRMQEXAMPLEMTSAMPLER2_H_ */

50
examples/MQ/multiple-transports/FairMQExampleMTSink.cxx Normal file
View File

@ -0,0 +1,50 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQExampleMTSink.cxx
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#include "FairMQExampleMTSink.h"
#include "FairMQLogger.h"
using namespace std;
FairMQExampleMTSink::FairMQExampleMTSink()
{
// register a handler for data arriving on "data" channel
OnData("data1", &FairMQExampleMTSink::HandleData1);
OnData("data2", &FairMQExampleMTSink::HandleData2);
}
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool FairMQExampleMTSink::HandleData1(FairMQMessagePtr& /*msg*/, int /*index*/)
{
// Creates a message using the transport of channel ack
FairMQMessagePtr ack(NewMessageFor("ack", 0));
if (Send(ack, "ack") < 0)
{
return false;
}
// return true if want to be called again (otherwise go to IDLE state)
return true;
}
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool FairMQExampleMTSink::HandleData2(FairMQMessagePtr& /*msg*/, int /*index*/)
{
// return true if want to be called again (otherwise go to IDLE state)
return true;
}
FairMQExampleMTSink::~FairMQExampleMTSink()
{
}

31
examples/MQ/multiple-transports/FairMQExampleMTSink.h Normal file
View File

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

12
examples/MQ/multiple-transports/README.md Normal file
View File

@ -0,0 +1,12 @@
Multiple Transports example
===========================
This example demonstrates use of multiple transports (zeromq/nanomsg/shmem) within the same topology and/or device. It is a simple topology consisting of two samplers and a sink. The devices are connected via 3 channels:
![Multiple Transports example](../../../docs/images/fairmq-ex-multiple-transports.png?raw=true "Multiple Transports example")
Each device has main transport that it uses. By default it is ZeroMQ, and can be overriden via the `--transport` cmd option. The device will initialize additional transports if any of the channels have them configured (e.g. via the JSON file, see `ex-multiple-transports.json`).
In this example sampler1 and sink are started with `--transport shmem`, making shared memory their main transport, sampler2 with `--transport nanomsg`. Additionally, the ack channel is configured to use zeromq as its transport via the JSON configuration.
The main two things that a transport does is transfer of data and allocation of memory for the messages. By default, new messages are created via the main device transport. If a message has been created with one transport and is to be transferred with another, it has to be copied into a new message of the target transport. This happens automatically behind the scenes. To avoid this copy the device can create messages via `NewMessageFor(const string& channelName, int subChannelIndex, ...)` method, that creates the messages via the transport of the given channel (check sampler1 and sink for an example).

84
examples/MQ/multiple-transports/ex-multiple-transports.json Normal file
View File

@ -0,0 +1,84 @@
{
"fairMQOptions": {
"devices": [
{
"id": "sampler1",
"channels": [
{
"name": "data1",
"sockets": [
{
"type": "push",
"method": "bind",
"address": "tcp://127.0.0.1:5555"
}
]
},
{
"name": "ack",
"transport": "zeromq",
"sockets": [
{
"type": "sub",
"method": "bind",
"address": "tcp://127.0.0.1:5557"
}
]
}
]
},
{
"id": "sampler2",
"channels": [
{
"name": "data2",
"sockets": [
{
"type": "push",
"method": "bind",
"address": "tcp://127.0.0.1:5556"
}
]
}
]
},
{
"id": "sink1",
"channels": [
{
"name": "data1",
"sockets": [
{
"type": "pull",
"method": "connect",
"address": "tcp://127.0.0.1:5555"
}
]
},
{
"name": "data2",
"transport": "nanomsg",
"sockets": [
{
"type": "pull",
"method": "connect",
"address": "tcp://127.0.0.1:5556"
}
]
},
{
"name": "ack",
"transport": "zeromq",
"sockets": [
{
"type": "pub",
"method": "connect",
"address": "tcp://127.0.0.1:5557"
}
]
}
]
}
]
}
}

23
examples/MQ/multiple-transports/runExampleMTSampler1.cxx Normal file
View File

@ -0,0 +1,23 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "runFairMQDevice.h"
#include "FairMQExampleMTSampler1.h"
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& options)
{
options.add_options()
("text", bpo::value<std::string>()->default_value("Hello"), "Text to send out");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
{
return new FairMQExampleMTSampler1();
}

23
examples/MQ/multiple-transports/runExampleMTSampler2.cxx Normal file
View File

@ -0,0 +1,23 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "runFairMQDevice.h"
#include "FairMQExampleMTSampler2.h"
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& options)
{
options.add_options()
("text", bpo::value<std::string>()->default_value("Hello"), "Text to send out");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
{
return new FairMQExampleMTSampler2();
}

21
examples/MQ/multiple-transports/runExampleMTSink.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 version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "runFairMQDevice.h"
#include "FairMQExampleMTSink.h"
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& /*options*/)
{
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
{
return new FairMQExampleMTSink();
}

20
examples/MQ/multiple-transports/startMTEx.sh.in Executable file
View File

@ -0,0 +1,20 @@
#!/bin/bash
config="@CMAKE_BINARY_DIR@/bin/config/ex-multiple-transports.json"
SAMPLER1="ex-mt-sampler1"
SAMPLER1+=" --id sampler1"
SAMPLER1+=" --transport shmem"
SAMPLER1+=" --mq-config $config"
xterm -geometry 80x30+0+0 -hold -e @CMAKE_BINARY_DIR@/bin/examples/MQ/multiple-transports/$SAMPLER1 &
SAMPLER2="ex-mt-sampler2"
SAMPLER2+=" --id sampler2"
SAMPLER2+=" --transport nanomsg"
SAMPLER2+=" --mq-config $config"
xterm -geometry 80x30+0+450 -hold -e @CMAKE_BINARY_DIR@/bin/examples/MQ/multiple-transports/$SAMPLER2 &
SINK="ex-mt-sink"
SINK+=" --id sink1"
SINK+=" --transport shmem"
SINK+=" --mq-config $config"
xterm -geometry 80x30+500+0 -hold -e @CMAKE_BINARY_DIR@/bin/examples/MQ/multiple-transports/$SINK &