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

Rename /example to /examples and move MQ examples in it

Browse Source
This commit is contained in:
Alexey Rybalchenko
2015-11-09 10:36:24 +01:00
committed by Mohammad Al-Turany
parent 736d61830c
commit f0a878a43a
73 changed files with 4652 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,90 @@
################################################################################
# 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/2-sampler-processor-sink/ex2-sampler-processor-sink.json ${CMAKE_BINARY_DIR}/bin/config/ex2-sampler-processor-sink.json)
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/2-sampler-processor-sink
${CMAKE_CURRENT_BINARY_DIR}
)
Set(SYSTEM_INCLUDE_DIRECTORIES
${Boost_INCLUDE_DIR}
)
If(NANOMSG_FOUND)
Set(INCLUDE_DIRECTORIES
${INCLUDE_DIRECTORIES}
${CMAKE_SOURCE_DIR}/fairmq/nanomsg
)
Set(SYSTEM_INCLUDE_DIRECTORIES
${SYSTEM_INCLUDE_DIRECTORIES}
${ZMQ_INCLUDE_DIR}
)
Else(NANOMSG_FOUND)
Set(INCLUDE_DIRECTORIES
${INCLUDE_DIRECTORIES}
${CMAKE_SOURCE_DIR}/fairmq/zeromq
)
Set(SYSTEM_INCLUDE_DIRECTORIES
${SYSTEM_INCLUDE_DIRECTORIES}
${ZMQ_INCLUDE_DIR}
)
EndIf(NANOMSG_FOUND)
Include_Directories(${INCLUDE_DIRECTORIES})
Include_Directories(SYSTEM ${SYSTEM_INCLUDE_DIRECTORIES})
Set(LINK_DIRECTORIES
${Boost_LIBRARY_DIRS}
)
Link_Directories(${LINK_DIRECTORIES})
Set(SRCS
"FairMQExample2Sampler.cxx"
"FairMQExample2Processor.cxx"
"FairMQExample2Sink.cxx"
)
Set(DEPENDENCIES
${DEPENDENCIES}
FairMQ
)
Set(LIBRARY_NAME FairMQExample2)
GENERATE_LIBRARY()
Set(Exe_Names
ex2-sampler
ex2-processor
ex2-sink
)
Set(Exe_Source
runExample2Sampler.cxx
runExample2Processor.cxx
runExample2Sink.cxx
)
list(LENGTH Exe_Names _length)
math(EXPR _length ${_length}-1)
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 FairMQExample2)
GENERATE_EXECUTABLE()
EndForEach(_file RANGE 0 ${_length})

62
examples/MQ/2-sampler-processor-sink/FairMQExample2Processor.cxx Normal file
View File

@@ -0,0 +1,62 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* FairMQExample2Processor.cpp
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include "FairMQExample2Processor.h"
#include "FairMQLogger.h"
using namespace std;
FairMQExample2Processor::FairMQExample2Processor()
: fText()
{
}
void FairMQExample2Processor::CustomCleanup(void *data, void *object)
{
delete (string*)object;
}
void FairMQExample2Processor::Run()
{
// Check if we are still in the RUNNING state
while (CheckCurrentState(RUNNING))
{
// Create empty message to hold the input
unique_ptr<FairMQMessage> input(fTransportFactory->CreateMessage());
// Receive the message (blocks until received or interrupted (e.g. by state change)).
// Returns size of the received message or -1 if interrupted.
if (fChannels.at("data-in").at(0).Receive(input) >= 0)
{
LOG(INFO) << "Received data, processing...";
// Modify the received string
string* text = new string(static_cast<char*>(input->GetData()), input->GetSize());
*text += " (modified by " + fId + ")";
// Create output message
unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text));
// Send out the output message
fChannels.at("data-out").at(0).Send(msg);
}
}
}
FairMQExample2Processor::~FairMQExample2Processor()
{
}

41
examples/MQ/2-sampler-processor-sink/FairMQExample2Processor.h 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 version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQExample2Processor.h
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#ifndef FAIRMQEXAMPLE2PROCESSOR_H_
#define FAIRMQEXAMPLE2PROCESSOR_H_
#include <string>
#include "FairMQDevice.h"
class FairMQExample2Processor : public FairMQDevice
{
public:
enum
{
Text = FairMQDevice::Last,
Last
};
FairMQExample2Processor();
virtual ~FairMQExample2Processor();
static void CustomCleanup(void* data, void* hint);
protected:
std::string fText;
virtual void Run();
};
#endif /* FAIRMQEXAMPLE2PROCESSOR_H_ */

96
examples/MQ/2-sampler-processor-sink/FairMQExample2Sampler.cxx Normal file
View File

@@ -0,0 +1,96 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* FairMQExample2Sampler.cpp
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include "FairMQExample2Sampler.h"
#include "FairMQLogger.h"
using namespace std;
FairMQExample2Sampler::FairMQExample2Sampler()
: fText()
{
}
void FairMQExample2Sampler::CustomCleanup(void *data, void *object)
{
delete (string*)object;
}
void FairMQExample2Sampler::Run()
{
// Check if we are still in the RUNNING state
while (CheckCurrentState(RUNNING))
{
boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
string* text = new string(fText);
unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text));
LOG(INFO) << "Sending \"" << fText << "\"";
fChannels.at("data-out").at(0).Send(msg);
}
}
FairMQExample2Sampler::~FairMQExample2Sampler()
{
}
void FairMQExample2Sampler::SetProperty(const int key, const string& value)
{
switch (key)
{
case Text:
fText = value;
break;
default:
FairMQDevice::SetProperty(key, value);
break;
}
}
string FairMQExample2Sampler::GetProperty(const int key, const string& default_ /*= ""*/)
{
switch (key)
{
case Text:
return fText;
break;
default:
return FairMQDevice::GetProperty(key, default_);
}
}
void FairMQExample2Sampler::SetProperty(const int key, const int value)
{
switch (key)
{
default:
FairMQDevice::SetProperty(key, value);
break;
}
}
int FairMQExample2Sampler::GetProperty(const int key, const int default_ /*= 0*/)
{
switch (key)
{
default:
return FairMQDevice::GetProperty(key, default_);
}
}

46
examples/MQ/2-sampler-processor-sink/FairMQExample2Sampler.h Normal file
View File

@@ -0,0 +1,46 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* FairMQExample2Sampler.h
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#ifndef FAIRMQEXAMPLE2SAMPLER_H_
#define FAIRMQEXAMPLE2SAMPLER_H_
#include <string>
#include "FairMQDevice.h"
class FairMQExample2Sampler : public FairMQDevice
{
public:
enum
{
Text = FairMQDevice::Last,
Last
};
FairMQExample2Sampler();
virtual ~FairMQExample2Sampler();
static void CustomCleanup(void* data, void* hint);
virtual void SetProperty(const int key, const std::string& value);
virtual std::string GetProperty(const int key, const std::string& default_ = "");
virtual void SetProperty(const int key, const int value);
virtual int GetProperty(const int key, const int default_ = 0);
protected:
std::string fText;
virtual void Run();
};
#endif /* FAIRMQEXAMPLE2SAMPLER_H_ */

44
examples/MQ/2-sampler-processor-sink/FairMQExample2Sink.cxx Normal file
View File

@@ -0,0 +1,44 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* FairMQExample2Sink.cxx
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include "FairMQExample2Sink.h"
#include "FairMQLogger.h"
using namespace std;
FairMQExample2Sink::FairMQExample2Sink()
{
}
void FairMQExample2Sink::Run()
{
while (CheckCurrentState(RUNNING))
{
unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
if (fChannels.at("data-in").at(0).Receive(msg) >= 0)
{
LOG(INFO) << "Received message: \""
<< string(static_cast<char*>(msg->GetData()), msg->GetSize())
<< "\"";
}
}
}
FairMQExample2Sink::~FairMQExample2Sink()
{
}

30
examples/MQ/2-sampler-processor-sink/FairMQExample2Sink.h Normal file
View File

@@ -0,0 +1,30 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* FairMQExample2Sink.h
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#ifndef FAIRMQEXAMPLE2SINK_H_
#define FAIRMQEXAMPLE2SINK_H_
#include "FairMQDevice.h"
class FairMQExample2Sink : public FairMQDevice
{
public:
FairMQExample2Sink();
virtual ~FairMQExample2Sink();
protected:
virtual void Run();
};
#endif /* FAIRMQEXAMPLE2SINK_H_ */

10
examples/MQ/2-sampler-processor-sink/README.md Normal file
View File

@@ -0,0 +1,10 @@
Example 2: Sampler -> Processor -> Sink
===============
A simple topology of three devices - **Sampler**, **Processor** and **Sink**. **Sampler** sends data to one or more **Processor**s, who modify the data and send it to one **Sink**. Transport with the **PUSH-PULL** pattern.
In this example the Sampler is configured to **bind** its output and the Sink is configured to also **bind** its input. This allows us run any number of processors with the same configuration, because they all connect to same Sampler and Sink addresses. Furthermore, it allows adding of processors dynamically during run-time. The PUSH and PULL sockets will handle the data distribution to/from the new devices according to their distribution strategies ([Round-robin output for PUSH](http://api.zeromq.org/4-0:zmq-socket#toc14) and [Fair-queued input for PULL](http://api.zeromq.org/4-0:zmq-socket#toc15)).
The Sampler sends out a simple text string (its content configurable with `--text` command line parameter, defaul is "Hello"). Each Processor modifies the string by appending its ID to it and send it to the Sink.
The provided configuration file contains two Processors. To add more Processors, you can either extend the configuration file, or create a separate file only for new processors.

102
examples/MQ/2-sampler-processor-sink/ex2-sampler-processor-sink.json Normal file
View File

@@ -0,0 +1,102 @@
{
"fairMQOptions":
{
"device":
{
"id": "sampler1",
"channel":
{
"name": "data-out",
"socket":
{
"type": "push",
"method": "bind",
"address": "tcp://*:5555",
"sndBufSize": "1000",
"rcvBufSize": "1000",
"rateLogging": "0"
}
}
},
"device":
{
"id": "processor1",
"channel":
{
"name": "data-in",
"socket":
{
"type": "pull",
"method": "connect",
"address": "tcp://localhost:5555",
"sndBufSize": "1000",
"rcvBufSize": "1000",
"rateLogging": "0"
}
},
"channel":
{
"name": "data-out",
"socket":
{
"type": "push",
"method": "connect",
"address": "tcp://localhost:5556",
"sndBufSize": "1000",
"rcvBufSize": "1000",
"rateLogging": "0"
}
}
},
"device":
{
"id": "processor2",
"channel":
{
"name": "data-in",
"socket":
{
"type": "pull",
"method": "connect",
"address": "tcp://localhost:5555",
"sndBufSize": "1000",
"rcvBufSize": "1000",
"rateLogging": "0"
}
},
"channel":
{
"name": "data-out",
"socket":
{
"type": "push",
"method": "connect",
"address": "tcp://localhost:5556",
"sndBufSize": "1000",
"rcvBufSize": "1000",
"rateLogging": "0"
}
}
},
"device":
{
"id": "sink1",
"channel":
{
"name": "data-in",
"socket":
{
"type": "pull",
"method": "bind",
"address": "tcp://*:5556",
"sndBufSize": "1000",
"rcvBufSize": "1000",
"rateLogging": "0"
}
}
}
}
}

79
examples/MQ/2-sampler-processor-sink/runExample2Processor.cxx Normal file
View File

@@ -0,0 +1,79 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* runExample2Processor.cxx
*
* @since 2013-04-23
* @author D. Klein, A. Rybalchenko
*/
#include <iostream>
#include "FairMQLogger.h"
#include "FairMQParser.h"
#include "FairMQProgOptions.h"
#include "FairMQExample2Processor.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#endif
int main(int argc, char** argv)
{
FairMQExample2Processor processor;
processor.CatchSignals();
FairMQProgOptions config;
try
{
if (config.ParseAll(argc, argv))
{
return 0;
}
std::string filename = config.GetValue<std::string>("config-json-file");
std::string id = config.GetValue<std::string>("id");
config.UserParser<FairMQParser::JSON>(filename, id);
processor.fChannels = config.GetFairMQMap();
LOG(INFO) << "PID: " << getpid();
#ifdef NANOMSG
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
#else
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
#endif
processor.SetTransport(transportFactory);
processor.SetProperty(FairMQExample2Processor::Id, id);
processor.ChangeState("INIT_DEVICE");
processor.WaitForEndOfState("INIT_DEVICE");
processor.ChangeState("INIT_TASK");
processor.WaitForEndOfState("INIT_TASK");
processor.ChangeState("RUN");
processor.InteractiveStateLoop();
}
catch (std::exception& e)
{
LOG(ERROR) << e.what();
LOG(INFO) << "Command line options are the following: ";
config.PrintHelp();
return 1;
}
return 0;
}

92
examples/MQ/2-sampler-processor-sink/runExample2Sampler.cxx Normal file
View File

@@ -0,0 +1,92 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* runExample2Sampler.cxx
*
* @since 2013-04-23
* @author D. Klein, A. Rybalchenko
*/
#include <iostream>
#include "boost/program_options.hpp"
#include "FairMQLogger.h"
#include "FairMQParser.h"
#include "FairMQProgOptions.h"
#include "FairMQExample2Sampler.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#endif
using namespace boost::program_options;
int main(int argc, char** argv)
{
FairMQExample2Sampler sampler;
sampler.CatchSignals();
FairMQProgOptions config;
try
{
std::string text;
options_description samplerOptions("Sampler options");
samplerOptions.add_options()
("text", value<std::string>(&text)->default_value("Hello"), "Text to send out");
config.AddToCmdLineOptions(samplerOptions);
if (config.ParseAll(argc, argv))
{
return 0;
}
std::string filename = config.GetValue<std::string>("config-json-file");
std::string id = config.GetValue<std::string>("id");
config.UserParser<FairMQParser::JSON>(filename, id);
sampler.fChannels = config.GetFairMQMap();
LOG(INFO) << "PID: " << getpid();
#ifdef NANOMSG
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
#else
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
#endif
sampler.SetTransport(transportFactory);
sampler.SetProperty(FairMQExample2Sampler::Id, id);
sampler.SetProperty(FairMQExample2Sampler::Text, text);
sampler.ChangeState("INIT_DEVICE");
sampler.WaitForEndOfState("INIT_DEVICE");
sampler.ChangeState("INIT_TASK");
sampler.WaitForEndOfState("INIT_TASK");
sampler.ChangeState("RUN");
sampler.InteractiveStateLoop();
}
catch (std::exception& e)
{
LOG(ERROR) << e.what();
LOG(INFO) << "Command line options are the following: ";
config.PrintHelp();
return 1;
}
return 0;
}

79
examples/MQ/2-sampler-processor-sink/runExample2Sink.cxx Normal file
View File

@@ -0,0 +1,79 @@
/********************************************************************************
* 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" *
********************************************************************************/
/**
* runExample2Sink.cxx
*
* @since 2013-04-23
* @author D. Klein, A. Rybalchenko
*/
#include <iostream>
#include "FairMQLogger.h"
#include "FairMQParser.h"
#include "FairMQProgOptions.h"
#include "FairMQExample2Sink.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#endif
int main(int argc, char** argv)
{
FairMQExample2Sink sink;
sink.CatchSignals();
FairMQProgOptions config;
try
{
if (config.ParseAll(argc, argv))
{
return 0;
}
std::string filename = config.GetValue<std::string>("config-json-file");
std::string id = config.GetValue<std::string>("id");
config.UserParser<FairMQParser::JSON>(filename, id);
sink.fChannels = config.GetFairMQMap();
LOG(INFO) << "PID: " << getpid();
#ifdef NANOMSG
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
#else
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
#endif
sink.SetTransport(transportFactory);
sink.SetProperty(FairMQExample2Sink::Id, id);
sink.ChangeState("INIT_DEVICE");
sink.WaitForEndOfState("INIT_DEVICE");
sink.ChangeState("INIT_TASK");
sink.WaitForEndOfState("INIT_TASK");
sink.ChangeState("RUN");
sink.InteractiveStateLoop();
}
catch (std::exception& e)
{
LOG(ERROR) << e.what();
LOG(INFO) << "Command line options are the following: ";
config.PrintHelp();
return 1;
}
return 0;
}