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

use clang-format for FairMQ

Browse Source
This commit is contained in:
Alexey Rybalchenko 2014-04-10 15:20:48 +02:00
parent e80e6d4269
commit 68d51d8ed5
55 changed files with 1893 additions and 1759 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
fairmq/.clang-format Normal file
View File

@ -0,0 +1,54 @@
---
Language: Cpp
#AccessModifierOffset: -4
ConstructorInitializerIndentWidth: 4
AlignEscapedNewlinesLeft: true
AlignTrailingComments: true
AllowAllParametersOfDeclarationOnNextLine: false
AllowShortIfStatementsOnASingleLine: false
AllowShortLoopsOnASingleLine: false
AllowShortFunctionsOnASingleLine: false
AlwaysBreakTemplateDeclarations: true
# It is broken on windows. Breaks all #include "header.h"
#AlwaysBreakBeforeMultilineStrings: true
BreakBeforeBinaryOperators: false
BreakBeforeTernaryOperators: true
BreakConstructorInitializersBeforeComma: true
BinPackParameters: false
ColumnLimit: 160
ConstructorInitializerAllOnOneLineOrOnePerLine: false
ConstructorInitializerIndentWidth: 4
DerivePointerBinding: false
ExperimentalAutoDetectBinPacking: false
IndentCaseLabels: true
MaxEmptyLinesToKeep: 1
NamespaceIndentation: All
ObjCSpaceAfterProperty: false
ObjCSpaceBeforeProtocolList: false
PenaltyBreakBeforeFirstCallParameter: 1
PenaltyBreakComment: 300
PenaltyBreakString: 1000
PenaltyBreakFirstLessLess: 120
PenaltyExcessCharacter: 1000000
PenaltyReturnTypeOnItsOwnLine: 200
PointerBindsToType: true
SpacesBeforeTrailingComments: 1
Cpp11BracedListStyle: false
Standard: Cpp11
IndentWidth: 4
TabWidth: 4
UseTab: Never
BreakBeforeBraces: Allman
IndentFunctionDeclarationAfterType: true
SpacesInParentheses: false
SpacesInAngles: false
SpaceInEmptyParentheses: false
SpacesInCStyleCastParentheses: false
SpacesInContainerLiterals: true
SpaceBeforeAssignmentOperators: true
ContinuationIndentWidth: 4
CommentPragmas: '^ IWYU pragma:'
SpaceBeforeParens: ControlStatements
...

54
fairmq/FairMQBenchmarkSampler.cxx
View File

@ -13,11 +13,10 @@
#include "FairMQBenchmarkSampler.h"
#include "FairMQLogger.h"
FairMQBenchmarkSampler::FairMQBenchmarkSampler() :
fEventSize(10000),
fEventRate(1),
fEventCounter(0)
FairMQBenchmarkSampler::FairMQBenchmarkSampler()
: fEventSize(10000)
, fEventRate(1)
, fEventCounter(0)
{
}
@ -33,7 +32,7 @@ void FairMQBenchmarkSampler::Init()
void FairMQBenchmarkSampler::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
//boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
// boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
boost::thread resetEventCounter(boost::bind(&FairMQBenchmarkSampler::ResetEventCounter, this));
@ -41,7 +40,8 @@ void FairMQBenchmarkSampler::Run()
void* buffer = operator new[](fEventSize);
FairMQMessage* base_msg = fTransportFactory->CreateMessage(buffer, fEventSize);
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
msg->Copy(base_msg);
@ -49,7 +49,8 @@ void FairMQBenchmarkSampler::Run()
--fEventCounter;
while (fEventCounter == 0) {
while (fEventCounter == 0)
{
boost::this_thread::sleep(boost::posix_time::milliseconds(1));
}
@ -67,11 +68,15 @@ void FairMQBenchmarkSampler::Run()
void FairMQBenchmarkSampler::ResetEventCounter()
{
while ( true ) {
try {
while (true)
{
try
{
fEventCounter = fEventRate / 100;
boost::this_thread::sleep(boost::posix_time::milliseconds(10));
} catch (boost::thread_interrupted&) {
}
catch (boost::thread_interrupted&)
{
break;
}
}
@ -90,7 +95,8 @@ void FairMQBenchmarkSampler::Log(int intervalInMs)
t0 = get_timestamp();
while (true) {
while (true)
{
boost::this_thread::sleep(boost::posix_time::milliseconds(intervalInMs));
t1 = get_timestamp();
@ -100,8 +106,8 @@ void FairMQBenchmarkSampler::Log(int intervalInMs)
timestamp_t timeSinceLastLog_ms = (t1 - t0) / 1000.0L;
megabytesPerSecond = ((double) (bytesNew - bytes) / (1024. * 1024.)) / (double) timeSinceLastLog_ms * 1000.;
messagesPerSecond = (double) (messagesNew - messages) / (double) timeSinceLastLog_ms * 1000.;
megabytesPerSecond = ((double)(bytesNew - bytes) / (1024. * 1024.)) / (double)timeSinceLastLog_ms * 1000.;
messagesPerSecond = (double)(messagesNew - messages) / (double)timeSinceLastLog_ms * 1000.;
LOG(DEBUG) << "send " << messagesPerSecond << " msg/s, " << megabytesPerSecond << " MB/s";
@ -111,26 +117,29 @@ void FairMQBenchmarkSampler::Log(int intervalInMs)
}
}
void FairMQBenchmarkSampler::SetProperty(const int key, const string& value, const int slot/*= 0*/)
void FairMQBenchmarkSampler::SetProperty(const int key, const string& value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
default:
FairMQDevice::SetProperty(key, value, slot);
break;
}
}
string FairMQBenchmarkSampler::GetProperty(const int key, const string& default_/*= ""*/, const int slot/*= 0*/)
string FairMQBenchmarkSampler::GetProperty(const int key, const string& default_ /*= ""*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
default:
return FairMQDevice::GetProperty(key, default_, slot);
}
}
void FairMQBenchmarkSampler::SetProperty(const int key, const int value, const int slot/*= 0*/)
void FairMQBenchmarkSampler::SetProperty(const int key, const int value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case EventSize:
fEventSize = value;
break;
@ -143,9 +152,10 @@ void FairMQBenchmarkSampler::SetProperty(const int key, const int value, const i
}
}
int FairMQBenchmarkSampler::GetProperty(const int key, const int default_/*= 0*/, const int slot/*= 0*/)
int FairMQBenchmarkSampler::GetProperty(const int key, const int default_ /*= 0*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case EventSize:
return fEventSize;
case EventRate:

5
fairmq/FairMQBenchmarkSampler.h
View File

@ -16,10 +16,11 @@
* Sampler to generate traffic for benchmarking.
*/
class FairMQBenchmarkSampler: public FairMQDevice
class FairMQBenchmarkSampler : public FairMQDevice
{
public:
enum {
enum
{
InputFile = FairMQDevice::Last,
EventRate,
EventSize,

8
fairmq/FairMQBuffer.cxx
View File

@ -13,7 +13,6 @@
#include "FairMQBuffer.h"
#include "FairMQLogger.h"
FairMQBuffer::FairMQBuffer()
{
}
@ -25,12 +24,14 @@ void FairMQBuffer::Run()
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
bool received = false;
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
if (received) {
if (received)
{
fPayloadOutputs->at(0)->Send(msg);
received = false;
}
@ -45,4 +46,3 @@ void FairMQBuffer::Run()
FairMQBuffer::~FairMQBuffer()
{
}

4
fairmq/FairMQBuffer.h
View File

@ -10,12 +10,12 @@
#include "FairMQDevice.h"
class FairMQBuffer: public FairMQDevice
class FairMQBuffer : public FairMQDevice
{
public:
FairMQBuffer();
virtual ~FairMQBuffer();
protected:
virtual void Run();
};

9
fairmq/FairMQConfigurable.cxx
View File

@ -7,25 +7,24 @@
#include "FairMQConfigurable.h"
FairMQConfigurable::FairMQConfigurable()
{
}
void FairMQConfigurable::SetProperty(const int key, const string& value, const int slot/*= 0*/)
void FairMQConfigurable::SetProperty(const int key, const string& value, const int slot /*= 0*/)
{
}
string FairMQConfigurable::GetProperty(const int key, const string& default_/*= ""*/, const int slot/*= 0*/)
string FairMQConfigurable::GetProperty(const int key, const string& default_ /*= ""*/, const int slot /*= 0*/)
{
return default_;
}
void FairMQConfigurable::SetProperty(const int key, const int value, const int slot/*= 0*/)
void FairMQConfigurable::SetProperty(const int key, const int value, const int slot /*= 0*/)
{
}
int FairMQConfigurable::GetProperty(const int key, const int default_/*= 0*/, const int slot/*= 0*/)
int FairMQConfigurable::GetProperty(const int key, const int default_ /*= 0*/, const int slot /*= 0*/)
{
return default_;
}

3
fairmq/FairMQConfigurable.h
View File

@ -15,7 +15,8 @@ using std::string;
class FairMQConfigurable
{
public:
enum {
enum
{
Last = 1
};
FairMQConfigurable();

80
fairmq/FairMQDevice.cxx
View File

@ -11,12 +11,13 @@
#include "FairMQDevice.h"
#include "FairMQLogger.h"
FairMQDevice::FairMQDevice() :
fNumIoThreads(1),
//fPayloadContext(NULL),
fPayloadInputs(new vector<FairMQSocket*>()),
fPayloadOutputs(new vector<FairMQSocket*>()),
fLogIntervalInMs(1000)
FairMQDevice::FairMQDevice()
: fNumIoThreads(1)
,
// fPayloadContext(NULL),
fPayloadInputs(new vector<FairMQSocket*>())
, fPayloadOutputs(new vector<FairMQSocket*>())
, fLogIntervalInMs(1000)
{
}
@ -72,7 +73,6 @@ void FairMQDevice::InitInput()
}
} catch (std::out_of_range& e) {
}
}
}
void FairMQDevice::InitOutput()
@ -95,7 +95,6 @@ void FairMQDevice::InitOutput()
}
} catch (std::out_of_range& e) {
}
}
}
void FairMQDevice::Run()
@ -107,9 +106,10 @@ void FairMQDevice::Pause()
}
// Method for setting properties represented as a string.
void FairMQDevice::SetProperty(const int key, const string& value, const int slot/*= 0*/)
void FairMQDevice::SetProperty(const int key, const string& value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case Id:
fId = value;
break;
@ -144,9 +144,10 @@ void FairMQDevice::SetProperty(const int key, const string& value, const int slo
}
// Method for setting properties represented as an integer.
void FairMQDevice::SetProperty(const int key, const int value, const int slot/*= 0*/)
void FairMQDevice::SetProperty(const int key, const int value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case NumIoThreads:
fNumIoThreads = value;
break;
@ -182,9 +183,10 @@ void FairMQDevice::SetProperty(const int key, const int value, const int slot/*=
}
// Method for getting properties represented as an string.
string FairMQDevice::GetProperty(const int key, const string& default_/*= ""*/, const int slot/*= 0*/)
string FairMQDevice::GetProperty(const int key, const string& default_ /*= ""*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case Id:
return fId;
case InputAddress:
@ -205,9 +207,10 @@ string FairMQDevice::GetProperty(const int key, const string& default_/*= ""*/,
}
// Method for getting properties represented as an integer.
int FairMQDevice::GetProperty(const int key, const int default_/*= 0*/, const int slot/*= 0*/)
int FairMQDevice::GetProperty(const int key, const int default_ /*= 0*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case NumIoThreads:
return fNumIoThreads;
case LogIntervalInMs:
@ -260,14 +263,16 @@ void FairMQDevice::LogSocketRates()
// End of temp stuff
int i = 0;
for ( vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++ ) {
for (vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++)
{
bytesInput[i] = (*itr)->GetBytesRx();
messagesInput[i] = (*itr)->GetMessagesRx();
++i;
}
i = 0;
for ( vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++ ) {
for (vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++)
{
bytesOutput[i] = (*itr)->GetBytesTx();
messagesOutput[i] = (*itr)->GetMessagesTx();
++i;
@ -275,20 +280,23 @@ void FairMQDevice::LogSocketRates()
t0 = get_timestamp();
while ( true ) {
try {
while (true)
{
try
{
t1 = get_timestamp();
timeSinceLastLog_ms = (t1 - t0) / 1000.0L;
i = 0;
for ( vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++ ) {
for (vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++)
{
bytesInputNew[i] = (*itr)->GetBytesRx();
megabytesPerSecondInput[i] = ((double) (bytesInputNew[i] - bytesInput[i]) / (1024. * 1024.)) / (double) timeSinceLastLog_ms * 1000.;
megabytesPerSecondInput[i] = ((double)(bytesInputNew[i] - bytesInput[i]) / (1024. * 1024.)) / (double)timeSinceLastLog_ms * 1000.;
bytesInput[i] = bytesInputNew[i];
messagesInputNew[i] = (*itr)->GetMessagesRx();
messagesPerSecondInput[i] = (double) (messagesInputNew[i] - messagesInput[i]) / (double) timeSinceLastLog_ms * 1000.;
messagesPerSecondInput[i] = (double)(messagesInputNew[i] - messagesInput[i]) / (double)timeSinceLastLog_ms * 1000.;
messagesInput[i] = messagesInputNew[i];
LOG(DEBUG) << "#" << fId << "." << (*itr)->GetId() << ": " << messagesPerSecondInput[i] << " msg/s, " << megabytesPerSecondInput[i] << " MB/s";
@ -309,16 +317,17 @@ void FairMQDevice::LogSocketRates()
i = 0;
for ( vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++ )
for (vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++)
{
bytesOutputNew[i] = (*itr)->GetBytesTx();
megabytesPerSecondOutput[i] = ((double) (bytesOutputNew[i] - bytesOutput[i]) / (1024. * 1024.)) / (double) timeSinceLastLog_ms * 1000.;
megabytesPerSecondOutput[i] = ((double)(bytesOutputNew[i] - bytesOutput[i]) / (1024. * 1024.)) / (double)timeSinceLastLog_ms * 1000.;
bytesOutput[i] = bytesOutputNew[i];
messagesOutputNew[i] = (*itr)->GetMessagesTx();
messagesPerSecondOutput[i] = (double) (messagesOutputNew[i] - messagesOutput[i]) / (double) timeSinceLastLog_ms * 1000.;
messagesPerSecondOutput[i] = (double)(messagesOutputNew[i] - messagesOutput[i]) / (double)timeSinceLastLog_ms * 1000.;
messagesOutput[i] = messagesOutputNew[i];
LOG(DEBUG) << "#" << fId << "." << (*itr)->GetId() << ": " << messagesPerSecondOutput[i] << " msg/s, " << megabytesPerSecondOutput[i] << " MB/s";
LOG(DEBUG) << "#" << fId << "." << (*itr)->GetId() << ": " << messagesPerSecondOutput[i] << " msg/s, " << megabytesPerSecondOutput[i]
<< " MB/s";
// Temp stuff for process termination
// if ( !sentSomething && messagesPerSecondOutput[i] > 0 ) {
@ -347,7 +356,9 @@ void FairMQDevice::LogSocketRates()
t0 = t1;
boost::this_thread::sleep(boost::posix_time::milliseconds(fLogIntervalInMs));
} catch (boost::thread_interrupted&) {
}
catch (boost::thread_interrupted&)
{
cout << "rateLogger interrupted" << endl;
break;
}
@ -378,12 +389,14 @@ void FairMQDevice::ListenToCommands()
void FairMQDevice::Shutdown()
{
LOG(INFO) << ">>>>>>> closing inputs <<<<<<<";
for( vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++ ) {
for (vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++)
{
(*itr)->Close();
}
LOG(INFO) << ">>>>>>> closing outputs <<<<<<<";
for( vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++ ) {
for (vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++)
{
(*itr)->Close();
}
@ -393,11 +406,13 @@ void FairMQDevice::Shutdown()
FairMQDevice::~FairMQDevice()
{
for( vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++ ) {
for (vector<FairMQSocket*>::iterator itr = fPayloadInputs->begin(); itr != fPayloadInputs->end(); itr++)
{
delete (*itr);
}
for( vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++ ) {
for (vector<FairMQSocket*>::iterator itr = fPayloadOutputs->begin(); itr != fPayloadOutputs->end(); itr++)
{
delete (*itr);
}
@ -406,4 +421,3 @@ FairMQDevice::~FairMQDevice()
delete fPayloadInputs;
delete fPayloadOutputs;
}

27
fairmq/FairMQDevice.h
View File

@ -25,7 +25,8 @@ using std::endl;
class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
{
public:
enum {
enum
{
Id = FairMQConfigurable::Last,
NumIoThreads,
NumInputs,
@ -66,20 +67,20 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
int fNumInputs;
int fNumOutputs;
vector<string> *fInputAddress;
vector<string> *fInputMethod;
vector<string> *fInputSocketType;
vector<int> *fInputSndBufSize;
vector<int> *fInputRcvBufSize;
vector<string>* fInputAddress;
vector<string>* fInputMethod;
vector<string>* fInputSocketType;
vector<int>* fInputSndBufSize;
vector<int>* fInputRcvBufSize;
vector<string> *fOutputAddress;
vector<string> *fOutputMethod;
vector<string> *fOutputSocketType;
vector<int> *fOutputSndBufSize;
vector<int> *fOutputRcvBufSize;
vector<string>* fOutputAddress;
vector<string>* fOutputMethod;
vector<string>* fOutputSocketType;
vector<int>* fOutputSndBufSize;
vector<int>* fOutputRcvBufSize;
vector<FairMQSocket*> *fPayloadInputs;
vector<FairMQSocket*> *fPayloadOutputs;
vector<FairMQSocket*>* fPayloadInputs;
vector<FairMQSocket*>* fPayloadOutputs;
int fLogIntervalInMs;

11
fairmq/FairMQLogger.cxx
View File

@ -34,7 +34,8 @@ std::ostringstream& FairMQLogger::Log(int type)
std::strftime(mbstr, 100, "%H:%M:%S", std::localtime(&t));
string type_str;
switch (type) {
switch (type)
{
case DEBUG:
type_str = "\033[01;34mDEBUG\033[0m";
break;
@ -50,14 +51,16 @@ std::ostringstream& FairMQLogger::Log(int type)
break;
}
os << "[\033[01;36m" << mbstr << "\033[0m]" << "[" << type_str << "]" << " ";
os << "[\033[01;36m" << mbstr << "\033[0m]"
<< "[" << type_str << "]"
<< " ";
return os;
}
timestamp_t get_timestamp ()
timestamp_t get_timestamp()
{
struct timeval now;
gettimeofday (&now, NULL);
gettimeofday(&now, NULL);
return now.tv_usec + (timestamp_t)now.tv_sec * 1000000;
}

14
fairmq/FairMQLogger.h
View File

@ -21,21 +21,25 @@ using std::ostringstream;
class FairMQLogger
{
public:
enum {
DEBUG, INFO, ERROR, STATE
enum
{
DEBUG,
INFO,
ERROR,
STATE
};
FairMQLogger();
virtual ~FairMQLogger();
ostringstream& Log(int type);
private:
ostringstream os;
};
typedef unsigned long long timestamp_t;
timestamp_t get_timestamp ();
timestamp_t get_timestamp();
#define LOG(type) \
FairMQLogger().Log(FairMQLogger::type)
#define LOG(type) FairMQLogger().Log(FairMQLogger::type)
#endif /* FAIRMQLOGGER_H_ */

14
fairmq/FairMQMerger.cxx
View File

@ -12,7 +12,6 @@
#include "FairMQMerger.h"
#include "FairMQPoller.h"
FairMQMerger::FairMQMerger()
{
}
@ -31,16 +30,20 @@ void FairMQMerger::Run()
bool received = false;
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
poller->Poll(100);
for(int i = 0; i < fNumInputs; i++) {
if (poller->CheckInput(i)){
for (int i = 0; i < fNumInputs; i++)
{
if (poller->CheckInput(i))
{
received = fPayloadInputs->at(i)->Receive(msg);
}
if (received) {
if (received)
{
fPayloadOutputs->at(0)->Send(msg);
received = false;
}
@ -54,4 +57,3 @@ void FairMQMerger::Run()
rateLogger.interrupt();
rateLogger.join();
}

4
fairmq/FairMQMerger.h
View File

@ -10,12 +10,12 @@
#include "FairMQDevice.h"
class FairMQMerger: public FairMQDevice
class FairMQMerger : public FairMQDevice
{
public:
FairMQMerger();
virtual ~FairMQMerger();
protected:
virtual void Run();
};

1
fairmq/FairMQMessage.h
View File

@ -10,7 +10,6 @@
#include <cstddef> // for size_t
class FairMQMessage
{
public:

1
fairmq/FairMQPoller.h
View File

@ -15,7 +15,6 @@ class FairMQPoller
virtual bool CheckInput(int index) = 0;
virtual ~FairMQPoller() {};
};
#endif /* FAIRMQPOLLER_H_ */

3
fairmq/FairMQProxy.cxx
View File

@ -27,7 +27,8 @@ void FairMQProxy::Run()
FairMQMessage* msg = fTransportFactory->CreateMessage();
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
fPayloadInputs->at(0)->Receive(msg);
fPayloadOutputs->at(0)->Send(msg);
}

4
fairmq/FairMQProxy.h
View File

@ -10,12 +10,12 @@
#include "FairMQDevice.h"
class FairMQProxy: public FairMQDevice
class FairMQProxy : public FairMQDevice
{
public:
FairMQProxy();
virtual ~FairMQProxy();
protected:
virtual void Run();
};

4
fairmq/FairMQSink.cxx
View File

@ -21,7 +21,8 @@ void FairMQSink::Run()
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
fPayloadInputs->at(0)->Receive(msg);
@ -36,4 +37,3 @@ void FairMQSink::Run()
FairMQSink::~FairMQSink()
{
}

4
fairmq/FairMQSink.h
View File

@ -10,12 +10,12 @@
#include "FairMQDevice.h"
class FairMQSink: public FairMQDevice
class FairMQSink : public FairMQDevice
{
public:
FairMQSink();
virtual ~FairMQSink();
protected:
virtual void Run();
};

10
fairmq/FairMQSplitter.cxx
View File

@ -11,7 +11,6 @@
#include "FairMQLogger.h"
#include "FairMQSplitter.h"
FairMQSplitter::FairMQSplitter()
{
}
@ -29,15 +28,18 @@ void FairMQSplitter::Run()
bool received = false;
int direction = 0;
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
if (received) {
if (received)
{
fPayloadOutputs->at(direction)->Send(msg);
direction++;
if (direction >= fNumOutputs) {
if (direction >= fNumOutputs)
{
direction = 0;
}
received = false;

4
fairmq/FairMQSplitter.h
View File

@ -10,12 +10,12 @@
#include "FairMQDevice.h"
class FairMQSplitter: public FairMQDevice
class FairMQSplitter : public FairMQDevice
{
public:
FairMQSplitter();
virtual ~FairMQSplitter();
protected:
virtual void Run();
};

3
fairmq/apply-clang-format.sh Executable file
View File

@ -0,0 +1,3 @@
#!/bin/bash
find . -type f \( -iname "*.h" ! -iname "*.pb.h" -o -iname "*.cxx" \) -execdir clang-format -i {} \;

52
fairmq/nanomsg/FairMQMessageNN.cxx
View File

@ -12,17 +12,18 @@
#include "FairMQMessageNN.h"
#include "FairMQLogger.h"
FairMQMessageNN::FairMQMessageNN() :
fSize(0),
fMessage(NULL),
fReceiving(false)
FairMQMessageNN::FairMQMessageNN()
: fSize(0)
, fMessage(NULL)
, fReceiving(false)
{
}
FairMQMessageNN::FairMQMessageNN(size_t size)
{
fMessage = nn_allocmsg(size, 0);
if (!fMessage){
if (!fMessage)
{
LOG(ERROR) << "failed allocating message, reason: " << nn_strerror(errno);
}
fSize = size;
@ -32,10 +33,11 @@ FairMQMessageNN::FairMQMessageNN(size_t size)
FairMQMessageNN::FairMQMessageNN(void* data, size_t size)
{
fMessage = nn_allocmsg(size, 0);
if (!fMessage){
if (!fMessage)
{
LOG(ERROR) << "failed allocating message, reason: " << nn_strerror(errno);
}
memcpy (fMessage, data, size);
memcpy(fMessage, data, size);
fSize = size;
fReceiving = false;
}
@ -52,7 +54,8 @@ void FairMQMessageNN::Rebuild(size_t size)
{
Clear();
fMessage = nn_allocmsg(size, 0);
if (!fMessage){
if (!fMessage)
{
LOG(ERROR) << "failed allocating message, reason: " << nn_strerror(errno);
}
fSize = size;
@ -63,10 +66,11 @@ void FairMQMessageNN::Rebuild(void* data, size_t size)
{
Clear();
fMessage = nn_allocmsg(size, 0);
if (!fMessage){
if (!fMessage)
{
LOG(ERROR) << "failed allocating message, reason: " << nn_strerror(errno);
}
memcpy (fMessage, data, size);
memcpy(fMessage, data, size);
fSize = size;
fReceiving = false;
}
@ -94,9 +98,11 @@ void FairMQMessageNN::SetMessage(void* data, size_t size)
void FairMQMessageNN::Copy(FairMQMessage* msg)
{
if (fMessage){
if (fMessage)
{
int rc = nn_freemsg(fMessage);
if ( rc < 0 ){
if (rc < 0)
{
LOG(ERROR) << "failed freeing message, reason: " << nn_strerror(errno);
}
}
@ -104,19 +110,23 @@ void FairMQMessageNN::Copy(FairMQMessage* msg)
size_t size = msg->GetSize();
fMessage = nn_allocmsg(size, 0);
if (!fMessage){
if (!fMessage)
{
LOG(ERROR) << "failed allocating message, reason: " << nn_strerror(errno);
}
std::memcpy (fMessage, msg->GetMessage(), size);
std::memcpy(fMessage, msg->GetMessage(), size);
fSize = size;
}
inline void FairMQMessageNN::Clear()
{
int rc = nn_freemsg(fMessage);
if (rc < 0) {
if (rc < 0)
{
LOG(ERROR) << "failed freeing message, reason: " << nn_strerror(errno);
} else {
}
else
{
fMessage = NULL;
fSize = 0;
}
@ -124,11 +134,15 @@ inline void FairMQMessageNN::Clear()
FairMQMessageNN::~FairMQMessageNN()
{
if(fReceiving){
if (fReceiving)
{
int rc = nn_freemsg(fMessage);
if (rc < 0) {
if (rc < 0)
{
LOG(ERROR) << "failed freeing message, reason: " << nn_strerror(errno);
} else {
}
else
{
fMessage = NULL;
fSize = 0;
}

1
fairmq/nanomsg/FairMQMessageNN.h
View File

@ -12,7 +12,6 @@
#include "FairMQMessage.h"
class FairMQMessageNN : public FairMQMessage
{
public:

6
fairmq/nanomsg/FairMQPollerNN.cxx
View File

@ -14,7 +14,8 @@ FairMQPollerNN::FairMQPollerNN(const vector<FairMQSocket*>& inputs)
fNumItems = inputs.size();
items = new nn_pollfd[fNumItems];
for (int i = 0; i < fNumItems; i++) {
for (int i = 0; i < fNumItems; i++)
{
items[i].fd = inputs.at(i)->GetSocket(1);
items[i].events = NN_POLLIN;
}
@ -35,5 +36,6 @@ bool FairMQPollerNN::CheckInput(int index)
FairMQPollerNN::~FairMQPollerNN()
{
if (items != NULL) delete [] items;
if (items != NULL)
delete[] items;
}

47
fairmq/nanomsg/FairMQSocketNN.cxx
View File

@ -43,7 +43,8 @@ void FairMQSocketNN::Bind(const string& address)
LOG(INFO) << "bind socket #" << fId << " on " << address;
int eid = nn_bind(fSocket, address.c_str());
if (eid < 0) {
if (eid < 0)
{
LOG(ERROR) << "failed binding socket #" << fId << ", reason: " << nn_strerror(errno);
}
}
@ -53,7 +54,8 @@ void FairMQSocketNN::Connect(const string& address)
LOG(INFO) << "connect socket #" << fId << " to " << address;
int eid = nn_connect(fSocket, address.c_str());
if (eid < 0) {
if (eid < 0)
{
LOG(ERROR) << "failed connecting socket #" << fId << ", reason: " << nn_strerror(errno);
}
}
@ -62,9 +64,12 @@ size_t FairMQSocketNN::Send(FairMQMessage* msg)
{
void* ptr = msg->GetMessage();
int rc = nn_send(fSocket, &ptr, NN_MSG, 0);
if (rc < 0) {
if (rc < 0)
{
LOG(ERROR) << "failed sending on socket #" << fId << ", reason: " << nn_strerror(errno);
} else {
}
else
{
fBytesTx += rc;
++fMessagesTx;
static_cast<FairMQMessageNN*>(msg)->fReceiving = false;
@ -77,9 +82,12 @@ size_t FairMQSocketNN::Receive(FairMQMessage* msg)
{
void* ptr = NULL;
int rc = nn_recv(fSocket, &ptr, NN_MSG, 0);
if (rc < 0) {
if (rc < 0)
{
LOG(ERROR) << "failed receiving on socket #" << fId << ", reason: " << nn_strerror(errno);
} else {
}
else
{
fBytesRx += rc;
++fMessagesRx;
msg->SetMessage(ptr, rc);
@ -107,7 +115,8 @@ void FairMQSocketNN::Close()
void FairMQSocketNN::SetOption(const string& option, const void* value, size_t valueSize)
{
int rc = nn_setsockopt(fSocket, NN_SOL_SOCKET, GetConstant(option), value, valueSize);
if (rc < 0) {
if (rc < 0)
{
LOG(ERROR) << "failed setting socket option, reason: " << nn_strerror(errno);
}
}
@ -134,14 +143,22 @@ unsigned long FairMQSocketNN::GetMessagesRx()
int FairMQSocketNN::GetConstant(const string& constant)
{
if (constant == "sub") return NN_SUB;
if (constant == "pub") return NN_PUB;
if (constant == "xsub") return NN_SUB; // TODO: is there XPUB, XSUB for nanomsg?
if (constant == "xpub") return NN_PUB;
if (constant == "push") return NN_PUSH;
if (constant == "pull") return NN_PULL;
if (constant == "snd-hwm") return NN_SNDBUF;
if (constant == "rcv-hwm") return NN_RCVBUF;
if (constant == "sub")
return NN_SUB;
if (constant == "pub")
return NN_PUB;
if (constant == "xsub")
return NN_SUB; // TODO: is there XPUB, XSUB for nanomsg?
if (constant == "xpub")
return NN_PUB;
if (constant == "push")
return NN_PUSH;
if (constant == "pull")
return NN_PULL;
if (constant == "snd-hwm")
return NN_SNDBUF;
if (constant == "rcv-hwm")
return NN_RCVBUF;
return -1;
}

1
fairmq/nanomsg/FairMQSocketNN.h
View File

@ -14,7 +14,6 @@
#include "FairMQSocket.h"
class FairMQSocketNN : public FairMQSocket
{
public:

1
fairmq/nanomsg/FairMQTransportFactoryNN.h
View File

@ -26,7 +26,6 @@ class FairMQTransportFactoryNN : public FairMQTransportFactory
virtual FairMQSocket* CreateSocket(const string& type, int num, int numIoThreads);
virtual FairMQPoller* CreatePoller(const vector<FairMQSocket*>& inputs);
virtual ~FairMQTransportFactoryNN() {};
};

57
fairmq/prototest/FairMQBinSampler.cxx
View File

@ -15,11 +15,10 @@
#include "FairMQBinSampler.h"
#include "FairMQLogger.h"
FairMQBinSampler::FairMQBinSampler() :
fEventSize(10000),
fEventRate(1),
fEventCounter(0)
FairMQBinSampler::FairMQBinSampler()
: fEventSize(10000)
, fEventRate(1)
, fEventCounter(0)
{
}
@ -35,7 +34,7 @@ void FairMQBinSampler::Init()
void FairMQBinSampler::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
//boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
// boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
boost::thread resetEventCounter(boost::bind(&FairMQBinSampler::ResetEventCounter, this));
@ -44,11 +43,13 @@ void FairMQBinSampler::Run()
LOG(DEBUG) << "Message size: " << fEventSize * sizeof(Content) << " bytes.";
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
Content* payload = new Content[fEventSize];
for (int i = 0; i < fEventSize; ++i) {
for (int i = 0; i < fEventSize; ++i)
{
(&payload[i])->x = rand() % 100 + 1;
(&payload[i])->y = rand() % 100 + 1;
(&payload[i])->z = rand() % 100 + 1;
@ -64,7 +65,8 @@ void FairMQBinSampler::Run()
--fEventCounter;
while (fEventCounter == 0) {
while (fEventCounter == 0)
{
boost::this_thread::sleep(boost::posix_time::milliseconds(1));
}
@ -81,11 +83,15 @@ void FairMQBinSampler::Run()
void FairMQBinSampler::ResetEventCounter()
{
while ( true ) {
try {
while (true)
{
try
{
fEventCounter = fEventRate / 100;
boost::this_thread::sleep(boost::posix_time::milliseconds(10));
} catch (boost::thread_interrupted&) {
}
catch (boost::thread_interrupted&)
{
break;
}
}
@ -104,7 +110,8 @@ void FairMQBinSampler::Log(int intervalInMs)
t0 = get_timestamp();
while (true) {
while (true)
{
boost::this_thread::sleep(boost::posix_time::milliseconds(intervalInMs));
t1 = get_timestamp();
@ -114,8 +121,8 @@ void FairMQBinSampler::Log(int intervalInMs)
timestamp_t timeSinceLastLog_ms = (t1 - t0) / 1000.0L;
megabytesPerSecond = ((double) (bytesNew - bytes) / (1024. * 1024.)) / (double) timeSinceLastLog_ms * 1000.;
messagesPerSecond = (double) (messagesNew - messages) / (double) timeSinceLastLog_ms * 1000.;
megabytesPerSecond = ((double)(bytesNew - bytes) / (1024. * 1024.)) / (double)timeSinceLastLog_ms * 1000.;
messagesPerSecond = (double)(messagesNew - messages) / (double)timeSinceLastLog_ms * 1000.;
LOG(DEBUG) << "send " << messagesPerSecond << " msg/s, " << megabytesPerSecond << " MB/s";
@ -125,26 +132,29 @@ void FairMQBinSampler::Log(int intervalInMs)
}
}
void FairMQBinSampler::SetProperty(const int key, const string& value, const int slot/*= 0*/)
void FairMQBinSampler::SetProperty(const int key, const string& value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
default:
FairMQDevice::SetProperty(key, value, slot);
break;
}
}
string FairMQBinSampler::GetProperty(const int key, const string& default_/*= ""*/, const int slot/*= 0*/)
string FairMQBinSampler::GetProperty(const int key, const string& default_ /*= ""*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
default:
return FairMQDevice::GetProperty(key, default_, slot);
}
}
void FairMQBinSampler::SetProperty(const int key, const int value, const int slot/*= 0*/)
void FairMQBinSampler::SetProperty(const int key, const int value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case EventSize:
fEventSize = value;
break;
@ -157,9 +167,10 @@ void FairMQBinSampler::SetProperty(const int key, const int value, const int slo
}
}
int FairMQBinSampler::GetProperty(const int key, const int default_/*= 0*/, const int slot/*= 0*/)
int FairMQBinSampler::GetProperty(const int key, const int default_ /*= 0*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case EventSize:
return fEventSize;
case EventRate:

8
fairmq/prototest/FairMQBinSampler.h
View File

@ -12,7 +12,8 @@
#include "FairMQDevice.h"
struct Content {
struct Content
{
double a;
double b;
int x;
@ -20,10 +21,11 @@ struct Content {
int z;
};
class FairMQBinSampler: public FairMQDevice
class FairMQBinSampler : public FairMQDevice
{
public:
enum {
enum
{
InputFile = FairMQDevice::Last,
EventRate,
EventSize,

3
fairmq/prototest/FairMQBinSink.cxx
View File

@ -21,7 +21,8 @@ void FairMQBinSink::Run()
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
fPayloadInputs->at(0)->Receive(msg);

6
fairmq/prototest/FairMQBinSink.h
View File

@ -10,7 +10,8 @@
#include "FairMQDevice.h"
struct Content {
struct Content
{
double a;
double b;
int x;
@ -18,11 +19,12 @@ struct Content {
int z;
};
class FairMQBinSink: public FairMQDevice
class FairMQBinSink : public FairMQDevice
{
public:
FairMQBinSink();
virtual ~FairMQBinSink();
protected:
virtual void Run();
};

57
fairmq/prototest/FairMQProtoSampler.cxx
View File

@ -15,11 +15,10 @@
#include "payload.pb.h"
FairMQProtoSampler::FairMQProtoSampler() :
fEventSize(10000),
fEventRate(1),
fEventCounter(0)
FairMQProtoSampler::FairMQProtoSampler()
: fEventSize(10000)
, fEventRate(1)
, fEventCounter(0)
{
}
@ -35,18 +34,20 @@ void FairMQProtoSampler::Init()
void FairMQProtoSampler::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
//boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
// boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
boost::thread resetEventCounter(boost::bind(&FairMQProtoSampler::ResetEventCounter, this));
srand(time(NULL));
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
sampler::Payload p;
for (int i = 0; i < fEventSize; ++i) {
for (int i = 0; i < fEventSize; ++i)
{
sampler::Content* content = p.add_data();
content->set_x(rand() % 100 + 1);
@ -68,7 +69,8 @@ void FairMQProtoSampler::Run()
--fEventCounter;
while (fEventCounter == 0) {
while (fEventCounter == 0)
{
boost::this_thread::sleep(boost::posix_time::milliseconds(1));
}
@ -84,11 +86,15 @@ void FairMQProtoSampler::Run()
void FairMQProtoSampler::ResetEventCounter()
{
while ( true ) {
try {
while (true)
{
try
{
fEventCounter = fEventRate / 100;
boost::this_thread::sleep(boost::posix_time::milliseconds(10));
} catch (boost::thread_interrupted&) {
}
catch (boost::thread_interrupted&)
{
break;
}
}
@ -107,7 +113,8 @@ void FairMQProtoSampler::Log(int intervalInMs)
t0 = get_timestamp();
while (true) {
while (true)
{
boost::this_thread::sleep(boost::posix_time::milliseconds(intervalInMs));
t1 = get_timestamp();
@ -117,8 +124,8 @@ void FairMQProtoSampler::Log(int intervalInMs)
timestamp_t timeSinceLastLog_ms = (t1 - t0) / 1000.0L;
megabytesPerSecond = ((double) (bytesNew - bytes) / (1024. * 1024.)) / (double) timeSinceLastLog_ms * 1000.;
messagesPerSecond = (double) (messagesNew - messages) / (double) timeSinceLastLog_ms * 1000.;
megabytesPerSecond = ((double)(bytesNew - bytes) / (1024. * 1024.)) / (double)timeSinceLastLog_ms * 1000.;
messagesPerSecond = (double)(messagesNew - messages) / (double)timeSinceLastLog_ms * 1000.;
LOG(DEBUG) << "send " << messagesPerSecond << " msg/s, " << megabytesPerSecond << " MB/s";
@ -128,26 +135,29 @@ void FairMQProtoSampler::Log(int intervalInMs)
}
}
void FairMQProtoSampler::SetProperty(const int key, const string& value, const int slot/*= 0*/)
void FairMQProtoSampler::SetProperty(const int key, const string& value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
default:
FairMQDevice::SetProperty(key, value, slot);
break;
}
}
string FairMQProtoSampler::GetProperty(const int key, const string& default_/*= ""*/, const int slot/*= 0*/)
string FairMQProtoSampler::GetProperty(const int key, const string& default_ /*= ""*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
default:
return FairMQDevice::GetProperty(key, default_, slot);
}
}
void FairMQProtoSampler::SetProperty(const int key, const int value, const int slot/*= 0*/)
void FairMQProtoSampler::SetProperty(const int key, const int value, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case EventSize:
fEventSize = value;
break;
@ -160,9 +170,10 @@ void FairMQProtoSampler::SetProperty(const int key, const int value, const int s
}
}
int FairMQProtoSampler::GetProperty(const int key, const int default_/*= 0*/, const int slot/*= 0*/)
int FairMQProtoSampler::GetProperty(const int key, const int default_ /*= 0*/, const int slot /*= 0*/)
{
switch (key) {
switch (key)
{
case EventSize:
return fEventSize;
case EventRate:

5
fairmq/prototest/FairMQProtoSampler.h
View File

@ -12,10 +12,11 @@
#include "FairMQDevice.h"
class FairMQProtoSampler: public FairMQDevice
class FairMQProtoSampler : public FairMQDevice
{
public:
enum {
enum
{
InputFile = FairMQDevice::Last,
EventRate,
EventSize,

3
fairmq/prototest/FairMQProtoSink.cxx
View File

@ -23,7 +23,8 @@ void FairMQProtoSink::Run()
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
while ( fState == RUNNING ) {
while (fState == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
fPayloadInputs->at(0)->Receive(msg);

6
fairmq/prototest/FairMQProtoSink.h
View File

@ -10,7 +10,8 @@
#include "FairMQDevice.h"
struct Content {
struct Content
{
double a;
double b;
int x;
@ -18,11 +19,12 @@ struct Content {
int z;
};
class FairMQProtoSink: public FairMQDevice
class FairMQProtoSink : public FairMQDevice
{
public:
FairMQProtoSink();
virtual ~FairMQProtoSink();
protected:
virtual void Run();
};

21
fairmq/run/runBenchmarkSampler.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQBenchmarkSampler.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQBenchmarkSampler sampler;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,10 +47,10 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 9 ) {
if (argc != 9)
{
cout << "Usage: bsampler ID eventSize eventRate numIoTreads\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
<< endl;
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
return 1;
}
@ -90,10 +89,8 @@ int main(int argc, char** argv)
sampler.SetProperty(FairMQBenchmarkSampler::NumInputs, 0);
sampler.SetProperty(FairMQBenchmarkSampler::NumOutputs, 1);
sampler.ChangeState(FairMQBenchmarkSampler::INIT);
sampler.SetProperty(FairMQBenchmarkSampler::OutputSocketType, argv[i], 0);
++i;
int outputSndBufSize;
@ -105,13 +102,10 @@ int main(int argc, char** argv)
sampler.SetProperty(FairMQBenchmarkSampler::OutputAddress, argv[i], 0);
++i;
sampler.ChangeState(FairMQBenchmarkSampler::SETOUTPUT);
sampler.ChangeState(FairMQBenchmarkSampler::SETINPUT);
sampler.ChangeState(FairMQBenchmarkSampler::RUN);
char ch;
cin.get(ch);
@ -120,4 +114,3 @@ int main(int argc, char** argv)
return 0;
}

21
fairmq/run/runBinSampler.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQBinSampler.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQBinSampler sampler;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,10 +47,10 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 9 ) {
if (argc != 9)
{
cout << "Usage: bsampler ID eventSize eventRate numIoTreads\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
<< endl;
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
return 1;
}
@ -90,10 +89,8 @@ int main(int argc, char** argv)
sampler.SetProperty(FairMQBinSampler::NumInputs, 0);
sampler.SetProperty(FairMQBinSampler::NumOutputs, 1);
sampler.ChangeState(FairMQBinSampler::INIT);
sampler.SetProperty(FairMQBinSampler::OutputSocketType, argv[i], 0);
++i;
int outputSndBufSize;
@ -105,13 +102,10 @@ int main(int argc, char** argv)
sampler.SetProperty(FairMQBinSampler::OutputAddress, argv[i], 0);
++i;
sampler.ChangeState(FairMQBinSampler::SETOUTPUT);
sampler.ChangeState(FairMQBinSampler::SETINPUT);
sampler.ChangeState(FairMQBinSampler::RUN);
char ch;
cin.get(ch);
@ -120,4 +114,3 @@ int main(int argc, char** argv)
return 0;
}

20
fairmq/run/runBinSink.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQBinSink.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQBinSink sink;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,10 +47,10 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 7 ) {
if (argc != 7)
{
cout << "Usage: sink \tID numIoTreads\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< endl;
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n" << endl;
return 1;
}
@ -80,10 +79,8 @@ int main(int argc, char** argv)
sink.SetProperty(FairMQBinSink::NumInputs, 1);
sink.SetProperty(FairMQBinSink::NumOutputs, 0);
sink.ChangeState(FairMQBinSink::INIT);
sink.SetProperty(FairMQBinSink::InputSocketType, argv[i], 0);
++i;
int inputRcvBufSize;
@ -95,12 +92,10 @@ int main(int argc, char** argv)
sink.SetProperty(FairMQBinSink::InputAddress, argv[i], 0);
++i;
sink.ChangeState(FairMQBinSink::SETOUTPUT);
sink.ChangeState(FairMQBinSink::SETINPUT);
sink.ChangeState(FairMQBinSink::RUN);
char ch;
cin.get(ch);
@ -109,4 +104,3 @@ int main(int argc, char** argv)
return 0;
}

18
fairmq/run/runBuffer.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQBuffer.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQBuffer buffer;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,7 +47,8 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 11 ) {
if (argc != 11)
{
cout << "Usage: buffer \tID numIoTreads\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
@ -79,10 +79,8 @@ int main(int argc, char** argv)
buffer.SetProperty(FairMQBuffer::NumInputs, 1);
buffer.SetProperty(FairMQBuffer::NumOutputs, 1);
buffer.ChangeState(FairMQBuffer::INIT);
buffer.SetProperty(FairMQBuffer::InputSocketType, argv[i], 0);
++i;
int inputRcvBufSize;
@ -105,13 +103,10 @@ int main(int argc, char** argv)
buffer.SetProperty(FairMQBuffer::OutputAddress, argv[i], 0);
++i;
buffer.ChangeState(FairMQBuffer::SETOUTPUT);
buffer.ChangeState(FairMQBuffer::SETINPUT);
buffer.ChangeState(FairMQBuffer::RUN);
char ch;
cin.get(ch);
@ -120,4 +115,3 @@ int main(int argc, char** argv)
return 0;
}

20
fairmq/run/runMerger.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQMerger.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQMerger merger;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,13 +47,13 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc < 16 || (argc - 8) % 4 != 0 ) {
if (argc < 16 || (argc - 8) % 4 != 0)
{
cout << "Usage: merger \tID numIoTreads numInputs\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< "\t\t...\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
<< argc << " arguments provided" << endl;
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << argc << " arguments provided" << endl;
return 1;
}
@ -89,7 +88,8 @@ int main(int argc, char** argv)
merger.ChangeState(FairMQMerger::INIT);
for (int iInput = 0; iInput < numInputs; iInput++ ) {
for (int iInput = 0; iInput < numInputs; iInput++)
{
merger.SetProperty(FairMQMerger::InputSocketType, argv[i], iInput);
++i;
int inputRcvBufSize;
@ -117,7 +117,6 @@ int main(int argc, char** argv)
merger.ChangeState(FairMQMerger::SETINPUT);
merger.ChangeState(FairMQMerger::RUN);
char ch;
cin.get(ch);
@ -126,4 +125,3 @@ int main(int argc, char** argv)
return 0;
}

21
fairmq/run/runProtoSampler.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQProtoSampler.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQProtoSampler sampler;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,10 +47,10 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 9 ) {
if (argc != 9)
{
cout << "Usage: bsampler ID eventSize eventRate numIoTreads\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
<< endl;
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
return 1;
}
@ -90,10 +89,8 @@ int main(int argc, char** argv)
sampler.SetProperty(FairMQProtoSampler::NumInputs, 0);
sampler.SetProperty(FairMQProtoSampler::NumOutputs, 1);
sampler.ChangeState(FairMQProtoSampler::INIT);
sampler.SetProperty(FairMQProtoSampler::OutputSocketType, argv[i], 0);
++i;
int outputSndBufSize;
@ -105,13 +102,10 @@ int main(int argc, char** argv)
sampler.SetProperty(FairMQProtoSampler::OutputAddress, argv[i], 0);
++i;
sampler.ChangeState(FairMQProtoSampler::SETOUTPUT);
sampler.ChangeState(FairMQProtoSampler::SETINPUT);
sampler.ChangeState(FairMQProtoSampler::RUN);
char ch;
cin.get(ch);
@ -120,4 +114,3 @@ int main(int argc, char** argv)
return 0;
}

20
fairmq/run/runProtoSink.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQProtoSink.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQProtoSink sink;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,10 +47,10 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 7 ) {
if (argc != 7)
{
cout << "Usage: sink \tID numIoTreads\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< endl;
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n" << endl;
return 1;
}
@ -80,10 +79,8 @@ int main(int argc, char** argv)
sink.SetProperty(FairMQProtoSink::NumInputs, 1);
sink.SetProperty(FairMQProtoSink::NumOutputs, 0);
sink.ChangeState(FairMQProtoSink::INIT);
sink.SetProperty(FairMQProtoSink::InputSocketType, argv[i], 0);
++i;
int inputRcvBufSize;
@ -95,12 +92,10 @@ int main(int argc, char** argv)
sink.SetProperty(FairMQProtoSink::InputAddress, argv[i], 0);
++i;
sink.ChangeState(FairMQProtoSink::SETOUTPUT);
sink.ChangeState(FairMQProtoSink::SETINPUT);
sink.ChangeState(FairMQProtoSink::RUN);
char ch;
cin.get(ch);
@ -109,4 +104,3 @@ int main(int argc, char** argv)
return 0;
}

17
fairmq/run/runProxy.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQProxy.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQProxy proxy;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,7 +47,8 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 11 ) {
if (argc != 11)
{
cout << "Usage: proxy \tID numIoTreads\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
@ -80,10 +80,8 @@ int main(int argc, char** argv)
proxy.SetProperty(FairMQProxy::NumInputs, 1);
proxy.SetProperty(FairMQProxy::NumOutputs, 1);
proxy.ChangeState(FairMQProxy::INIT);
proxy.SetProperty(FairMQProxy::InputSocketType, argv[i], 0);
++i;
int inputRcvBufSize;
@ -106,12 +104,10 @@ int main(int argc, char** argv)
proxy.SetProperty(FairMQProxy::OutputAddress, argv[i], 0);
++i;
proxy.ChangeState(FairMQProxy::SETOUTPUT);
proxy.ChangeState(FairMQProxy::SETINPUT);
proxy.ChangeState(FairMQProxy::RUN);
char ch;
cin.get(ch);
@ -120,4 +116,3 @@ int main(int argc, char** argv)
return 0;
}

20
fairmq/run/runSink.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQSink.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQSink sink;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,10 +47,10 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc != 7 ) {
if (argc != 7)
{
cout << "Usage: sink \tID numIoTreads\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< endl;
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n" << endl;
return 1;
}
@ -80,10 +79,8 @@ int main(int argc, char** argv)
sink.SetProperty(FairMQSink::NumInputs, 1);
sink.SetProperty(FairMQSink::NumOutputs, 0);
sink.ChangeState(FairMQSink::INIT);
sink.SetProperty(FairMQSink::InputSocketType, argv[i], 0);
++i;
int inputRcvBufSize;
@ -95,12 +92,10 @@ int main(int argc, char** argv)
sink.SetProperty(FairMQSink::InputAddress, argv[i], 0);
++i;
sink.ChangeState(FairMQSink::SETOUTPUT);
sink.ChangeState(FairMQSink::SETINPUT);
sink.ChangeState(FairMQSink::RUN);
char ch;
cin.get(ch);
@ -109,4 +104,3 @@ int main(int argc, char** argv)
return 0;
}

20
fairmq/run/runSplitter.cxx
View File

@ -12,9 +12,9 @@
#include "FairMQSplitter.h"
#ifdef NANOMSG
#include "FairMQTransportFactoryNN.h"
#include "FairMQTransportFactoryNN.h"
#else
#include "FairMQTransportFactoryZMQ.h"
#include "FairMQTransportFactoryZMQ.h"
#endif
using std::cout;
@ -22,10 +22,9 @@ using std::cin;
using std::endl;
using std::stringstream;
FairMQSplitter splitter;
static void s_signal_handler (int signal)
static void s_signal_handler(int signal)
{
cout << endl << "Caught signal " << signal << endl;
@ -36,7 +35,7 @@ static void s_signal_handler (int signal)
exit(1);
}
static void s_catch_signals (void)
static void s_catch_signals(void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
@ -48,13 +47,13 @@ static void s_catch_signals (void)
int main(int argc, char** argv)
{
if ( argc < 16 || (argc - 8) % 4 != 0 ) {
if (argc < 16 || (argc - 8) % 4 != 0)
{
cout << "Usage: splitter \tID numIoTreads numOutputs\n"
<< "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
<< "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
<< "\t\t..."
<< argc << " arguments provided" << endl;
<< "\t\t..." << argc << " arguments provided" << endl;
return 1;
}
@ -101,7 +100,8 @@ int main(int argc, char** argv)
++i;
int outputSndBufSize;
for (int iOutput = 0; iOutput < numOutputs; iOutput++) {
for (int iOutput = 0; iOutput < numOutputs; iOutput++)
{
splitter.SetProperty(FairMQSplitter::OutputSocketType, argv[i], iOutput);
++i;
stringstream(argv[i]) >> outputSndBufSize;
@ -117,7 +117,6 @@ int main(int argc, char** argv)
splitter.ChangeState(FairMQSplitter::SETINPUT);
splitter.ChangeState(FairMQSplitter::RUN);
char ch;
cin.get(ch);
@ -126,4 +125,3 @@ int main(int argc, char** argv)
return 0;
}

18
fairmq/zeromq/FairMQContextZMQ.cxx
View File

@ -11,13 +11,15 @@
FairMQContextZMQ::FairMQContextZMQ(int numIoThreads)
{
fContext = zmq_ctx_new ();
if (fContext == NULL){
fContext = zmq_ctx_new();
if (fContext == NULL)
{
LOG(ERROR) << "failed creating context, reason: " << zmq_strerror(errno);
}
int rc = zmq_ctx_set (fContext, ZMQ_IO_THREADS, numIoThreads);
if (rc != 0){
int rc = zmq_ctx_set(fContext, ZMQ_IO_THREADS, numIoThreads);
if (rc != 0)
{
LOG(ERROR) << "failed configuring context, reason: " << zmq_strerror(errno);
}
}
@ -34,12 +36,14 @@ void* FairMQContextZMQ::GetContext()
void FairMQContextZMQ::Close()
{
if (fContext == NULL){
if (fContext == NULL)
{
return;
}
int rc = zmq_ctx_destroy (fContext);
if (rc != 0) {
int rc = zmq_ctx_destroy(fContext);
if (rc != 0)
{
LOG(ERROR) << "failed closing context, reason: " << zmq_strerror(errno);
}

48
fairmq/zeromq/FairMQMessageZMQ.cxx
View File

@ -11,27 +11,29 @@
#include "FairMQMessageZMQ.h"
#include "FairMQLogger.h"
FairMQMessageZMQ::FairMQMessageZMQ()
{
int rc = zmq_msg_init (&fMessage);
if (rc != 0) {
int rc = zmq_msg_init(&fMessage);
if (rc != 0)
{
LOG(ERROR) << "failed initializing message, reason: " << zmq_strerror(errno);
}
}
FairMQMessageZMQ::FairMQMessageZMQ(size_t size)
{
int rc = zmq_msg_init_size (&fMessage, size);
if (rc != 0) {
int rc = zmq_msg_init_size(&fMessage, size);
if (rc != 0)
{
LOG(ERROR) << "failed initializing message with size, reason: " << zmq_strerror(errno);
}
}
FairMQMessageZMQ::FairMQMessageZMQ(void* data, size_t size)
{
int rc = zmq_msg_init_data (&fMessage, data, size, &CleanUp, NULL);
if (rc != 0) {
int rc = zmq_msg_init_data(&fMessage, data, size, &CleanUp, NULL);
if (rc != 0)
{
LOG(ERROR) << "failed initializing message with data, reason: " << zmq_strerror(errno);
}
}
@ -39,8 +41,9 @@ FairMQMessageZMQ::FairMQMessageZMQ(void* data, size_t size)
void FairMQMessageZMQ::Rebuild()
{
CloseMessage();
int rc = zmq_msg_init (&fMessage);
if (rc != 0) {
int rc = zmq_msg_init(&fMessage);
if (rc != 0)
{
LOG(ERROR) << "failed initializing message, reason: " << zmq_strerror(errno);
}
}
@ -48,8 +51,9 @@ void FairMQMessageZMQ::Rebuild()
void FairMQMessageZMQ::Rebuild(size_t size)
{
CloseMessage();
int rc = zmq_msg_init_size (&fMessage, size);
if (rc != 0) {
int rc = zmq_msg_init_size(&fMessage, size);
if (rc != 0)
{
LOG(ERROR) << "failed initializing message with size, reason: " << zmq_strerror(errno);
}
}
@ -57,8 +61,9 @@ void FairMQMessageZMQ::Rebuild(size_t size)
void FairMQMessageZMQ::Rebuild(void* data, size_t size)
{
CloseMessage();
int rc = zmq_msg_init_data (&fMessage, data, size, &CleanUp, NULL);
if (rc != 0) {
int rc = zmq_msg_init_data(&fMessage, data, size, &CleanUp, NULL);
if (rc != 0)
{
LOG(ERROR) << "failed initializing message with data, reason: " << zmq_strerror(errno);
}
}
@ -70,12 +75,12 @@ void* FairMQMessageZMQ::GetMessage()
void* FairMQMessageZMQ::GetData()
{
return zmq_msg_data (&fMessage);
return zmq_msg_data(&fMessage);
}
size_t FairMQMessageZMQ::GetSize()
{
return zmq_msg_size (&fMessage);
return zmq_msg_size(&fMessage);
}
void FairMQMessageZMQ::SetMessage(void* data, size_t size)
@ -93,22 +98,23 @@ void FairMQMessageZMQ::Copy(FairMQMessage* msg)
inline void FairMQMessageZMQ::CloseMessage()
{
int rc = zmq_msg_close (&fMessage);
if (rc != 0) {
int rc = zmq_msg_close(&fMessage);
if (rc != 0)
{
LOG(ERROR) << "failed closing message, reason: " << zmq_strerror(errno);
}
}
void FairMQMessageZMQ::CleanUp(void* data, void* hint)
{
free (data);
free(data);
}
FairMQMessageZMQ::~FairMQMessageZMQ()
{
int rc = zmq_msg_close (&fMessage);
if (rc != 0) {
int rc = zmq_msg_close(&fMessage);
if (rc != 0)
{
LOG(ERROR) << "failed closing message with data, reason: " << zmq_strerror(errno);
}
}

1
fairmq/zeromq/FairMQMessageZMQ.h
View File

@ -14,7 +14,6 @@
#include "FairMQMessage.h"
class FairMQMessageZMQ : public FairMQMessage
{
public:

6
fairmq/zeromq/FairMQPollerZMQ.cxx
View File

@ -14,7 +14,8 @@ FairMQPollerZMQ::FairMQPollerZMQ(const vector<FairMQSocket*>& inputs)
fNumItems = inputs.size();
items = new zmq_pollitem_t[fNumItems];
for (int i = 0; i < fNumItems; i++) {
for (int i = 0; i < fNumItems; i++)
{
items[i].socket = inputs.at(i)->GetSocket();
items[i].fd = 0;
items[i].events = ZMQ_POLLIN;
@ -37,5 +38,6 @@ bool FairMQPollerZMQ::CheckInput(int index)
FairMQPollerZMQ::~FairMQPollerZMQ()
{
if (items != NULL) delete [] items;
if (items != NULL)
delete[] items;
}

69
fairmq/zeromq/FairMQSocketZMQ.cxx
View File

@ -39,6 +39,14 @@ FairMQSocketZMQ::FairMQSocketZMQ(const string& type, int num, int numIoThreads)
if (rc != 0) {
LOG(ERROR) << "failed setting socket option, reason: " << zmq_strerror(errno);
}
if (type == "sub")
{
rc = zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, NULL, 0);
if (rc != 0)
{
LOG(ERROR) << "failed setting socket option, reason: " << zmq_strerror(errno);
}
}
LOG(INFO) << "created socket #" << fId;
@ -53,8 +61,9 @@ void FairMQSocketZMQ::Bind(const string& address)
{
LOG(INFO) << "bind socket #" << fId << " on " << address;
int rc = zmq_bind (fSocket, address.c_str());
if (rc != 0) {
int rc = zmq_bind(fSocket, address.c_str());
if (rc != 0)
{
LOG(ERROR) << "failed binding socket #" << fId << ", reason: " << zmq_strerror(errno);
}
}
@ -63,21 +72,24 @@ void FairMQSocketZMQ::Connect(const string& address)
{
LOG(INFO) << "connect socket #" << fId << " on " << address;
int rc = zmq_connect (fSocket, address.c_str());
if (rc != 0) {
int rc = zmq_connect(fSocket, address.c_str());
if (rc != 0)
{
LOG(ERROR) << "failed connecting socket #" << fId << ", reason: " << zmq_strerror(errno);
}
}
size_t FairMQSocketZMQ::Send(FairMQMessage* msg)
{
int nbytes = zmq_msg_send (static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, 0);
if (nbytes >= 0){
int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, 0);
if (nbytes >= 0)
{
fBytesTx += nbytes;
++fMessagesTx;
return nbytes;
}
if (zmq_errno() == EAGAIN){
if (zmq_errno() == EAGAIN)
{
return false;
}
LOG(ERROR) << "failed sending on socket #" << fId << ", reason: " << zmq_strerror(errno);
@ -86,13 +98,15 @@ size_t FairMQSocketZMQ::Send(FairMQMessage* msg)
size_t FairMQSocketZMQ::Receive(FairMQMessage* msg)
{
int nbytes = zmq_msg_recv (static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, 0);
if (nbytes >= 0){
int nbytes = zmq_msg_recv(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, 0);
if (nbytes >= 0)
{
fBytesRx += nbytes;
++fMessagesRx;
return nbytes;
}
if (zmq_errno() == EAGAIN){
if (zmq_errno() == EAGAIN)
{
return false;
}
LOG(ERROR) << "failed receiving on socket #" << fId << ", reason: " << zmq_strerror(errno);
@ -101,12 +115,14 @@ size_t FairMQSocketZMQ::Receive(FairMQMessage* msg)
void FairMQSocketZMQ::Close()
{
if (fSocket == NULL){
if (fSocket == NULL)
{
return;
}
int rc = zmq_close (fSocket);
if (rc != 0) {
int rc = zmq_close(fSocket);
if (rc != 0)
{
LOG(ERROR) << "failed closing socket, reason: " << zmq_strerror(errno);
}
@ -127,7 +143,8 @@ int FairMQSocketZMQ::GetSocket(int nothing)
void FairMQSocketZMQ::SetOption(const string& option, const void* value, size_t valueSize)
{
int rc = zmq_setsockopt(fSocket, GetConstant(option), value, valueSize);
if (rc < 0) {
if (rc < 0)
{
LOG(ERROR) << "failed setting socket option, reason: " << zmq_strerror(errno);
}
}
@ -154,14 +171,22 @@ unsigned long FairMQSocketZMQ::GetMessagesRx()
int FairMQSocketZMQ::GetConstant(const string& constant)
{
if (constant == "sub") return ZMQ_SUB;
if (constant == "pub") return ZMQ_PUB;
if (constant == "xsub") return ZMQ_XSUB;
if (constant == "xpub") return ZMQ_XPUB;
if (constant == "push") return ZMQ_PUSH;
if (constant == "pull") return ZMQ_PULL;
if (constant == "snd-hwm") return ZMQ_SNDHWM;
if (constant == "rcv-hwm") return ZMQ_RCVHWM;
if (constant == "sub")
return ZMQ_SUB;
if (constant == "pub")
return ZMQ_PUB;
if (constant == "xsub")
return ZMQ_XSUB;
if (constant == "xpub")
return ZMQ_XPUB;
if (constant == "push")
return ZMQ_PUSH;
if (constant == "pull")
return ZMQ_PULL;
if (constant == "snd-hwm")
return ZMQ_SNDHWM;
if (constant == "rcv-hwm")
return ZMQ_RCVHWM;
return -1;
}

1
fairmq/zeromq/FairMQSocketZMQ.h
View File

@ -15,7 +15,6 @@
#include "FairMQSocket.h"
#include "FairMQContextZMQ.h"
class FairMQSocketZMQ : public FairMQSocket
{
public:

2
fairmq/zeromq/FairMQTransportFactoryZMQ.cxx
View File

@ -12,7 +12,7 @@
FairMQTransportFactoryZMQ::FairMQTransportFactoryZMQ()
{
int major, minor, patch;
zmq_version (&major, &minor, &patch);
zmq_version(&major, &minor, &patch);
LOG(INFO) << "Using ZeroMQ library, version: " << major << "." << minor << "." << patch;
}

1
fairmq/zeromq/FairMQTransportFactoryZMQ.h
View File

@ -27,7 +27,6 @@ class FairMQTransportFactoryZMQ : public FairMQTransportFactory
virtual FairMQSocket* CreateSocket(const string& type, int num, int numIoThreads);
virtual FairMQPoller* CreatePoller(const vector<FairMQSocket*>& inputs);
virtual ~FairMQTransportFactoryZMQ() {};
};