Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-12 16:21:13 +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 e53ad151a7
commit c66fd6fe91
39 changed files with 1840 additions and 1189 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
fairmq/CMakeLists.txt
View File

@ -111,6 +111,7 @@ EndIf(NANOMSG_FOUND)
# manual install (globbing add not recommended)
Set(FAIRMQHEADERS
FairMQParts.h
FairMQTransports.h
FairMQConfigPlugin.h
FairMQControlPlugin.h
runFairMQDevice.h

339
fairmq/FairMQChannel.cxx
View File

@ -13,11 +13,11 @@
*/
#include <set>
#include <utility> // std::move
#include <boost/algorithm/string.hpp> // join/split
#include "FairMQChannel.h"
#include "FairMQLogger.h"
using namespace std;
@ -30,18 +30,21 @@ FairMQChannel::FairMQChannel()
, fType("unspecified")
, fMethod("unspecified")
, fAddress("unspecified")
, fTransport("default")
, fSndBufSize(1000)
, fRcvBufSize(1000)
, fSndKernelSize(0)
, fRcvKernelSize(0)
, fRateLogging(1)
, fChannelName("")
, fName("")
, fIsValid(false)
, fPoller(nullptr)
, fCmdSocket(nullptr)
, fChannelCmdSocket(nullptr)
, fTransportType(FairMQ::Transport::DEFAULT)
, fTransportFactory(nullptr)
, fNoBlockFlag(0)
, fSndMoreFlag(0)
, fMultipart(false)
{
}
@ -50,18 +53,21 @@ FairMQChannel::FairMQChannel(const string& type, const string& method, const str
, fType(type)
, fMethod(method)
, fAddress(address)
, fTransport("default")
, fSndBufSize(1000)
, fRcvBufSize(1000)
, fSndKernelSize(0)
, fRcvKernelSize(0)
, fRateLogging(1)
, fChannelName("")
, fName("")
, fIsValid(false)
, fPoller(nullptr)
, fCmdSocket(nullptr)
, fChannelCmdSocket(nullptr)
, fTransportType(FairMQ::Transport::DEFAULT)
, fTransportFactory(nullptr)
, fNoBlockFlag(0)
, fSndMoreFlag(0)
, fMultipart(false)
{
}
@ -70,18 +76,21 @@ FairMQChannel::FairMQChannel(const FairMQChannel& chan)
, fType(chan.fType)
, fMethod(chan.fMethod)
, fAddress(chan.fAddress)
, fTransport(chan.fTransport)
, fSndBufSize(chan.fSndBufSize)
, fRcvBufSize(chan.fRcvBufSize)
, fSndKernelSize(chan.fSndKernelSize)
, fRcvKernelSize(chan.fRcvKernelSize)
, fRateLogging(chan.fRateLogging)
, fChannelName(chan.fChannelName)
, fName(chan.fName)
, fIsValid(false)
, fPoller(nullptr)
, fCmdSocket(nullptr)
, fChannelCmdSocket(nullptr)
, fTransportType(FairMQ::Transport::DEFAULT)
, fTransportFactory(nullptr)
, fNoBlockFlag(chan.fNoBlockFlag)
, fSndMoreFlag(chan.fSndMoreFlag)
, fMultipart(chan.fMultipart)
{}
FairMQChannel& FairMQChannel::operator=(const FairMQChannel& chan)
@ -89,16 +98,18 @@ FairMQChannel& FairMQChannel::operator=(const FairMQChannel& chan)
fType = chan.fType;
fMethod = chan.fMethod;
fAddress = chan.fAddress;
fTransport = chan.fTransport;
fSndBufSize = chan.fSndBufSize;
fRcvBufSize = chan.fRcvBufSize;
fSndKernelSize = chan.fSndKernelSize;
fRcvKernelSize = chan.fRcvKernelSize;
fRateLogging = chan.fRateLogging;
fSocket = nullptr;
fChannelName = chan.fChannelName;
fName = chan.fName;
fIsValid = false;
fPoller = nullptr;
fCmdSocket = nullptr;
fChannelCmdSocket = nullptr;
fTransportType = FairMQ::Transport::DEFAULT;
fTransportFactory = nullptr;
fNoBlockFlag = chan.fNoBlockFlag;
fSndMoreFlag = chan.fSndMoreFlag;
@ -108,13 +119,13 @@ FairMQChannel& FairMQChannel::operator=(const FairMQChannel& chan)
string FairMQChannel::GetChannelName() const
{
return fChannelName;
return fName;
}
string FairMQChannel::GetChannelPrefix() const
{
string prefix = fChannelName;
return prefix.erase(fChannelName.rfind("["));
string prefix = fName;
return prefix.erase(fName.rfind("["));
}
string FairMQChannel::GetType() const
@ -159,6 +170,20 @@ string FairMQChannel::GetAddress() const
}
}
string FairMQChannel::GetTransport() const
{
try
{
unique_lock<mutex> lock(fChannelMutex);
return fTransport;
}
catch (exception& e)
{
LOG(ERROR) << "Exception caught in FairMQChannel::GetTransport: " << e.what();
exit(EXIT_FAILURE);
}
}
int FairMQChannel::GetSndBufSize() const
{
try
@ -274,6 +299,21 @@ void FairMQChannel::UpdateAddress(const string& address)
}
}
void FairMQChannel::UpdateTransport(const string& transport)
{
try
{
unique_lock<mutex> lock(fChannelMutex);
fIsValid = false;
fTransport = transport;
}
catch (exception& e)
{
LOG(ERROR) << "Exception caught in FairMQChannel::UpdateTransport: " << e.what();
exit(EXIT_FAILURE);
}
}
void FairMQChannel::UpdateSndBufSize(const int sndBufSize)
{
try
@ -370,7 +410,7 @@ bool FairMQChannel::ValidateChannel()
unique_lock<mutex> lock(fChannelMutex);
stringstream ss;
ss << "Validating channel \"" << fChannelName << "\"... ";
ss << "Validating channel \"" << fName << "\"... ";
if (fIsValid)
{
@ -461,6 +501,17 @@ bool FairMQChannel::ValidateChannel()
}
}
// validate channel transport
// const string channelTransportNames[] = { "default", "zeromq", "nanomsg", "shmem" };
// const set<string> channelTransports(channelTransportNames, channelTransportNames + sizeof(channelTransportNames) / sizeof(string));
if (FairMQ::TransportTypes.find(fTransport) == FairMQ::TransportTypes.end())
{
ss << "INVALID";
LOG(DEBUG) << ss.str();
LOG(ERROR) << "Invalid channel transport: \"" << fTransport << "\"";
exit(EXIT_FAILURE);
}
// validate socket buffer size for sending
if (fSndBufSize < 0)
{
@ -518,19 +569,23 @@ bool FairMQChannel::ValidateChannel()
}
}
bool FairMQChannel::InitCommandInterface(shared_ptr<FairMQTransportFactory> factory, int numIoThreads)
void FairMQChannel::InitTransport(shared_ptr<FairMQTransportFactory> factory)
{
fTransportFactory = factory;
fTransportType = factory->GetType();
}
fCmdSocket = fTransportFactory->CreateSocket("sub", "device-commands", numIoThreads, "internal");
if (fCmdSocket)
bool FairMQChannel::InitCommandInterface(int numIoThreads)
{
fChannelCmdSocket = fTransportFactory->CreateSocket("sub", "device-commands", numIoThreads, "internal");
if (fChannelCmdSocket)
{
fCmdSocket->Connect("inproc://commands");
fChannelCmdSocket->Connect("inproc://commands");
fNoBlockFlag = fCmdSocket->NOBLOCK;
fSndMoreFlag = fCmdSocket->SNDMORE;
fNoBlockFlag = fChannelCmdSocket->NOBLOCK;
fSndMoreFlag = fChannelCmdSocket->SNDMORE;
fPoller = fTransportFactory->CreatePoller(*fCmdSocket, *fSocket);
fPoller = fTransportFactory->CreatePoller(*fChannelCmdSocket, *fSocket);
return true;
}
@ -547,7 +602,19 @@ void FairMQChannel::ResetChannel()
// TODO: implement channel resetting
}
int FairMQChannel::Send(const unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs) const
int FairMQChannel::Send(unique_ptr<FairMQMessage>& msg) const
{
CheckCompatibility(msg);
return fSocket->Send(msg);
}
int FairMQChannel::Receive(unique_ptr<FairMQMessage>& msg) const
{
CheckCompatibility(msg);
return fSocket->Receive(msg);
}
int FairMQChannel::Send(unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs) const
{
fPoller->Poll(sndTimeoutInMs);
@ -562,13 +629,13 @@ int FairMQChannel::Send(const unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs
if (fPoller->CheckOutput(1))
{
return fSocket->Send(msg.get(), 0);
return Send(msg);
}
return -2;
}
int FairMQChannel::Receive(const unique_ptr<FairMQMessage>& msg, int rcvTimeoutInMs) const
int FairMQChannel::Receive(unique_ptr<FairMQMessage>& msg, int rcvTimeoutInMs) const
{
fPoller->Poll(rcvTimeoutInMs);
@ -583,13 +650,37 @@ int FairMQChannel::Receive(const unique_ptr<FairMQMessage>& msg, int rcvTimeoutI
if (fPoller->CheckInput(1))
{
return fSocket->Receive(msg.get(), 0);
return Receive(msg);
}
return -2;
}
int64_t FairMQChannel::Send(const vector<unique_ptr<FairMQMessage>>& msgVec, int sndTimeoutInMs) const
int FairMQChannel::SendAsync(unique_ptr<FairMQMessage>& msg) const
{
CheckCompatibility(msg);
return fSocket->Send(msg, fNoBlockFlag);
}
int FairMQChannel::ReceiveAsync(unique_ptr<FairMQMessage>& msg) const
{
CheckCompatibility(msg);
return fSocket->Receive(msg, fNoBlockFlag);
}
int64_t FairMQChannel::Send(vector<unique_ptr<FairMQMessage>>& msgVec) const
{
CheckCompatibility(msgVec);
return fSocket->Send(msgVec);
}
int64_t FairMQChannel::Receive(vector<unique_ptr<FairMQMessage>>& msgVec) const
{
CheckCompatibility(msgVec);
return fSocket->Receive(msgVec);
}
int64_t FairMQChannel::Send(vector<unique_ptr<FairMQMessage>>& msgVec, int sndTimeoutInMs) const
{
fPoller->Poll(sndTimeoutInMs);
@ -604,7 +695,7 @@ int64_t FairMQChannel::Send(const vector<unique_ptr<FairMQMessage>>& msgVec, int
if (fPoller->CheckOutput(1))
{
return fSocket->Send(msgVec);
return Send(msgVec);
}
return -2;
@ -625,155 +716,33 @@ int64_t FairMQChannel::Receive(vector<unique_ptr<FairMQMessage>>& msgVec, int rc
if (fPoller->CheckInput(1))
{
return fSocket->Receive(msgVec);
return Receive(msgVec);
}
return -2;
}
int FairMQChannel::Send(FairMQMessage* msg, const string& flag, int sndTimeoutInMs) const
int64_t FairMQChannel::SendAsync(vector<unique_ptr<FairMQMessage>>& msgVec) const
{
if (flag == "")
{
fPoller->Poll(sndTimeoutInMs);
if (fPoller->CheckInput(0))
{
HandleUnblock();
if (fInterrupted)
{
return -2;
}
}
if (fPoller->CheckOutput(1))
{
return fSocket->Send(msg, flag);
}
return -2;
}
else
{
return fSocket->Send(msg, flag);
}
CheckCompatibility(msgVec);
return fSocket->Send(msgVec, fNoBlockFlag);
}
int FairMQChannel::Send(FairMQMessage* msg, const int flags, int sndTimeoutInMs) const
/// Receives a vector of messages in non-blocking mode.
///
/// @param msgVec message vector reference
/// @return Number of bytes that have been received. If queue is empty, returns -2.
/// In case of errors, returns -1.
int64_t FairMQChannel::ReceiveAsync(vector<unique_ptr<FairMQMessage>>& msgVec) const
{
if (flags == 0)
{
fPoller->Poll(sndTimeoutInMs);
if (fPoller->CheckInput(0))
{
HandleUnblock();
if (fInterrupted)
{
return -2;
}
}
if (fPoller->CheckOutput(1))
{
return fSocket->Send(msg, flags);
}
return -2;
}
else
{
return fSocket->Send(msg, flags);
}
}
int FairMQChannel::Receive(FairMQMessage* msg, const string& flag, int rcvTimeoutInMs) const
{
if (flag == "")
{
fPoller->Poll(rcvTimeoutInMs);
if (fPoller->CheckInput(0))
{
HandleUnblock();
if (fInterrupted)
{
return -2;
}
}
if (fPoller->CheckInput(1))
{
return fSocket->Receive(msg, flag);
}
return -2;
}
else
{
return fSocket->Receive(msg, flag);
}
}
int FairMQChannel::Receive(FairMQMessage* msg, const int flags, int rcvTimeoutInMs) const
{
if (flags == 0)
{
fPoller->Poll(rcvTimeoutInMs);
if (fPoller->CheckInput(0))
{
HandleUnblock();
if (fInterrupted)
{
return -2;
}
}
if (fPoller->CheckInput(1))
{
return fSocket->Receive(msg, flags);
}
return -2;
}
else
{
return fSocket->Receive(msg, flags);
}
}
bool FairMQChannel::ExpectsAnotherPart() const
{
int64_t more = 0;
size_t more_size = sizeof more;
if (fSocket)
{
fSocket->GetOption("rcv-more", &more, &more_size);
if (more)
{
return true;
}
else
{
return false;
}
}
else
{
return false;
}
CheckCompatibility(msgVec);
return fSocket->Receive(msgVec, fNoBlockFlag);
}
inline bool FairMQChannel::HandleUnblock() const
{
FairMQMessagePtr cmd(fTransportFactory->CreateMessage());
if (fCmdSocket->Receive(cmd.get(), 0) >= 0)
if (fChannelCmdSocket->Receive(cmd) >= 0)
{
// LOG(DEBUG) << "unblocked";
}
@ -788,3 +757,51 @@ void FairMQChannel::Tokenize(vector<string>& output, const string& input, const
{
boost::algorithm::split(output, input, boost::algorithm::is_any_of(delimiters));
}
FairMQTransportFactory* FairMQChannel::Transport()
{
return fTransportFactory.get();
}
bool FairMQChannel::CheckCompatibility(unique_ptr<FairMQMessage>& msg) const
{
if (fTransportType == msg->GetType())
{
return true;
}
else
{
// LOG(WARN) << "Channel type does not match message type. Copying...";
FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage(msg->GetSize()));
memcpy(msgCopy->GetData(), msg->GetData(), msg->GetSize());
msg = move(msgCopy);
return false;
}
}
bool FairMQChannel::CheckCompatibility(vector<unique_ptr<FairMQMessage>>& msgVec) const
{
if (msgVec.size() > 0)
{
if (fTransportType == msgVec.at(0)->GetType())
{
return true;
}
else
{
// LOG(WARN) << "Channel type does not match message type. Copying...";
vector<unique_ptr<FairMQMessage>> tempVec;
for (unsigned int i = 0; i < msgVec.size(); ++i)
{
tempVec.push_back(fTransportFactory->CreateMessage(msgVec[i]->GetSize()));
memcpy(tempVec[i]->GetData(), msgVec[i]->GetData(), msgVec[i]->GetSize());
}
msgVec = move(tempVec);
return false;
}
}
else
{
return true;
}
}

131
fairmq/FairMQChannel.h
View File

@ -17,12 +17,15 @@
#include <string>
#include <memory> // unique_ptr
#include <vector>
#include <atomic>
#include <mutex>
#include "FairMQTransportFactory.h"
#include "FairMQSocket.h"
#include "FairMQPoller.h"
#include "FairMQTransports.h"
#include "FairMQLogger.h"
class FairMQPoller;
class FairMQTransportFactory;
@ -67,9 +70,13 @@ class FairMQChannel
std::string GetMethod() const;
/// Get socket address (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")
/// @return Returns socket type (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")
/// @return Returns socket address (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")
std::string GetAddress() const;
/// Get channel transport ("default", "zeromq", "nanomsg" or "shmem")
/// @return Returns channel transport (e.g. "default", "zeromq", "nanomsg" or "shmem")
std::string GetTransport() const;
/// Get socket send buffer size (in number of messages)
/// @return Returns socket send buffer size (in number of messages)
int GetSndBufSize() const;
@ -99,9 +106,13 @@ class FairMQChannel
void UpdateMethod(const std::string& method);
/// Set socket address
/// @param address Socket address (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")
/// @param Socket address (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")
void UpdateAddress(const std::string& address);
/// Set channel transport
/// @param transport transport string ("default", "zeromq", "nanomsg" or "shmem")
void UpdateTransport(const std::string& transport);
/// Set socket send buffer size
/// @param sndBufSize Socket send buffer size (in number of messages)
void UpdateSndBufSize(const int sndBufSize);
@ -135,6 +146,9 @@ class FairMQChannel
std::unique_ptr<FairMQSocket> fSocket;
int Send(std::unique_ptr<FairMQMessage>& msg) const;
int Receive(std::unique_ptr<FairMQMessage>& msg) const;
/// Sends a message to the socket queue.
/// @details Send method attempts to send a message by
/// putting it in the output queue. If the queue is full or queueing is not possible
@ -143,7 +157,16 @@ class FairMQChannel
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
int Send(const std::unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs = -1) const;
int Send(std::unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs) const;
/// Receives a message from the socket queue.
/// @details Receive method attempts to receive a message from the input queue.
/// If the queue is empty the method blocks.
///
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @return Number of bytes that have been received. -2 If reading from the queue was not possible or timed out.
/// In case of errors, returns -1.
int Receive(std::unique_ptr<FairMQMessage>& msg, int rcvTimeoutInMs) const;
/// Sends a message in non-blocking mode.
/// @details SendAsync method attempts to send a message without blocking by
@ -153,41 +176,31 @@ class FairMQChannel
/// @return Number of bytes that have been queued. If queueing failed due to
/// full queue or no connected peers (when binding), returns -2.
/// In case of errors, returns -1.
inline int SendAsync(const std::unique_ptr<FairMQMessage>& msg) const
{
return fSocket->Send(msg.get(), fNoBlockFlag);
}
int SendAsync(std::unique_ptr<FairMQMessage>& msg) const;
/// Queues the current message as a part of a multi-part message
/// @details SendPart method queues the provided message as a part of a multi-part message.
/// The actual transfer over the network is initiated once final part has been queued with the Send() or SendAsync() methods.
/// Receives a message in non-blocking mode.
///
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @return Number of bytes that have been queued. -2 If queueing was not possible.
/// @return Number of bytes that have been received. If queue is empty, returns -2.
/// In case of errors, returns -1.
inline int SendPart(const std::unique_ptr<FairMQMessage>& msg) const
{
return fSocket->Send(msg.get(), fSndMoreFlag);
}
int ReceiveAsync(std::unique_ptr<FairMQMessage>& msg) const;
/// Queues the current message as a part of a multi-part message without blocking
/// @details SendPart method queues the provided message as a part of a multi-part message without blocking.
/// The actual transfer over the network is initiated once final part has been queued with the Send() or SendAsync() methods.
///
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @return Number of bytes that have been queued. -2 If queueing was not possible.
/// In case of errors, returns -1.
inline int SendPartAsync(const std::unique_ptr<FairMQMessage>& msg) const
{
return fSocket->Send(msg.get(), fSndMoreFlag|fNoBlockFlag);
}
int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;
int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;
/// Send a vector of messages
///
/// @param msgVec message vector reference
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
int64_t Send(const std::vector<std::unique_ptr<FairMQMessage>>& msgVec, int sndTimeoutInMs = -1) const;
int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, int sndTimeoutInMs) const;
/// Receive a vector of messages
///
/// @param msgVec message vector reference
/// @return Number of bytes that have been received. -2 If reading from the queue was not possible or timed out.
/// In case of errors, returns -1.
int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, int rcvTimeoutInMs) const;
/// Sends a vector of message in non-blocking mode.
/// @details SendAsync method attempts to send a vector of messages without blocking by
@ -196,82 +209,49 @@ class FairMQChannel
/// @param msgVec message vector reference
/// @return Number of bytes that have been queued. If queueing failed due to
/// full queue or no connected peers (when binding), returns -2. In case of errors, returns -1.
inline int64_t SendAsync(const std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const
{
return fSocket->Send(msgVec, fNoBlockFlag);
}
/// Receives a message from the socket queue.
/// @details Receive method attempts to receive a message from the input queue.
/// If the queue is empty the method blocks.
///
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @return Number of bytes that have been received. -2 If reading from the queue was not possible or timed out.
/// In case of errors, returns -1.
int Receive(const std::unique_ptr<FairMQMessage>& msg, int rcvTimeoutInMs = -1) const;
/// Receives a message in non-blocking mode.
///
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @return Number of bytes that have been received. If queue is empty, returns -2.
/// In case of errors, returns -1.
inline int ReceiveAsync(const std::unique_ptr<FairMQMessage>& msg) const
{
return fSocket->Receive(msg.get(), fNoBlockFlag);
}
/// Receive a vector of messages
///
/// @param msgVec message vector reference
/// @return Number of bytes that have been received. -2 If reading from the queue was not possible or timed out.
/// In case of errors, returns -1.
int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, int rcvTimeoutInMs = -1) const;
int64_t SendAsync(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;
/// Receives a vector of messages in non-blocking mode.
///
/// @param msgVec message vector reference
/// @return Number of bytes that have been received. If queue is empty, returns -2.
/// In case of errors, returns -1.
inline int64_t ReceiveAsync(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const
{
return fSocket->Receive(msgVec, fNoBlockFlag);
}
/// Checks if the socket is expecting to receive another part of a multipart message.
/// @return Return true if the socket expects another part of a multipart message and false otherwise.
bool ExpectsAnotherPart() const;
// DEPRECATED socket method wrappers with raw pointers and flag checks
int Send(FairMQMessage* msg, const std::string& flag = "", int sndTimeoutInMs = -1) const;
int Send(FairMQMessage* msg, const int flags, int sndTimeoutInMs = -1) const;
int Receive(FairMQMessage* msg, const std::string& flag = "", int rcvTimeoutInMs = -1) const;
int Receive(FairMQMessage* msg, const int flags, int rcvTimeoutInMs = -1) const;
int64_t ReceiveAsync(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;
// TODO: this might go to some base utility library
static void Tokenize(std::vector<std::string>& output, const std::string& input, const std::string delimiters = ",");
FairMQTransportFactory* Transport();
private:
std::string fType;
std::string fMethod;
std::string fAddress;
std::string fTransport;
int fSndBufSize;
int fRcvBufSize;
int fSndKernelSize;
int fRcvKernelSize;
int fRateLogging;
std::string fChannelName;
std::string fName;
std::atomic<bool> fIsValid;
FairMQPollerPtr fPoller;
FairMQSocketPtr fCmdSocket;
FairMQPollerPtr fPoller;
FairMQSocketPtr fChannelCmdSocket;
FairMQ::Transport fTransportType;
std::shared_ptr<FairMQTransportFactory> fTransportFactory;
int fNoBlockFlag;
int fSndMoreFlag;
bool InitCommandInterface(std::shared_ptr<FairMQTransportFactory> factory, int numIoThreads);
bool CheckCompatibility(std::unique_ptr<FairMQMessage>& msg) const;
bool CheckCompatibility(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;
void InitTransport(std::shared_ptr<FairMQTransportFactory> factory);
bool InitCommandInterface(int numIoThreads);
bool HandleUnblock() const;
@ -282,6 +262,7 @@ class FairMQChannel
static std::mutex fChannelMutex;
static std::atomic<bool> fInterrupted;
bool fMultipart;
};
#endif /* FAIRMQCHANNEL_H_ */

724
fairmq/FairMQDevice.cxx
View File

File diff suppressed because it is too large Load Diff

185
fairmq/FairMQDevice.h
View File

@ -17,6 +17,7 @@
#include <vector>
#include <memory> // unique_ptr
#include <algorithm> // std::sort()
#include <string>
#include <iostream>
#include <unordered_map>
@ -30,6 +31,7 @@
#include "FairMQConfigurable.h"
#include "FairMQStateMachine.h"
#include "FairMQTransportFactory.h"
#include "FairMQTransports.h"
#include "FairMQSocket.h"
#include "FairMQChannel.h"
@ -96,83 +98,103 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
Deserializer().Deserialize(msg, std::forward<DataType>(data), std::forward<Args>(args)...);
}
inline int Send(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).Send(msg);
}
inline int Receive(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).Receive(msg);
}
/// Shorthand method to send `msg` on `chan` at index `i`
/// @param msg message reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
inline int Send(const FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1) const
inline int Send(FairMQMessagePtr& msg, const std::string& chan, const int i, int sndTimeoutInMs) const
{
return fChannels.at(chan).at(i).Send(msg, sndTimeoutInMs);
}
/// Shorthand method to send `msg` on `chan` at index `i` without blocking
/// @param msg message reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
inline int SendAsync(const FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).SendAsync(msg);
}
/// Shorthand method to send FairMQParts on `chan` at index `i`
/// @param parts parts reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
inline int64_t Send(const FairMQParts& parts, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1) const
{
return fChannels.at(chan).at(i).Send(parts.fParts, sndTimeoutInMs);
}
/// Shorthand method to send FairMQParts on `chan` at index `i` without blocking
/// @param parts parts reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
inline int64_t SendAsync(const FairMQParts& parts, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).SendAsync(parts.fParts);
}
/// Shorthand method to receive `msg` on `chan` at index `i`
/// @param msg message reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been received. -2 If reading from the queue was not possible or timed out.
/// In case of errors, returns -1.
inline int Receive(const FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1) const
inline int Receive(FairMQMessagePtr& msg, const std::string& chan, const int i, int rcvTimeoutInMs) const
{
return fChannels.at(chan).at(i).Receive(msg, rcvTimeoutInMs);
}
/// Shorthand method to send `msg` on `chan` at index `i` without blocking
/// @param msg message reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
inline int SendAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).SendAsync(msg);
}
/// Shorthand method to receive `msg` on `chan` at index `i` without blocking
/// @param msg message reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been received. -2 If reading from the queue was not possible or timed out.
/// In case of errors, returns -1.
inline int ReceiveAsync(const FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const
inline int ReceiveAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).ReceiveAsync(msg);
}
inline int64_t Send(FairMQParts& parts, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).Send(parts.fParts);
}
inline int64_t Receive(FairMQParts& parts, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).Receive(parts.fParts);
}
/// Shorthand method to send FairMQParts on `chan` at index `i`
/// @param parts parts reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
inline int64_t Send(FairMQParts& parts, const std::string& chan, const int i, int sndTimeoutInMs) const
{
return fChannels.at(chan).at(i).Send(parts.fParts, sndTimeoutInMs);
}
/// Shorthand method to receive FairMQParts on `chan` at index `i`
/// @param parts parts reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been received. -2 If reading from the queue was not possible or timed out.
/// In case of errors, returns -1.
inline int64_t Receive(FairMQParts& parts, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1) const
inline int64_t Receive(FairMQParts& parts, const std::string& chan, const int i, int rcvTimeoutInMs) const
{
return fChannels.at(chan).at(i).Receive(parts.fParts, rcvTimeoutInMs);
}
/// Shorthand method to send FairMQParts on `chan` at index `i` without blocking
/// @param parts parts reference
/// @param chan channel name
/// @param i channel index
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out.
/// In case of errors, returns -1.
inline int64_t SendAsync(FairMQParts& parts, const std::string& chan, const int i = 0) const
{
return fChannels.at(chan).at(i).SendAsync(parts.fParts);
}
/// Shorthand method to receive FairMQParts on `chan` at index `i` without blocking
/// @param parts parts reference
/// @param chan channel name
@ -199,25 +221,25 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
return fTransportFactory->CreateMessage(size);
}
template<typename T>
static void FairMQSimpleMsgCleanup(void* /*data*/, void* hint)
{
delete static_cast<T*>(hint);
}
static void FairMQNoCleanup(void* /*data*/, void* /*hint*/)
{
}
/// @brief Create new FairMQMessage with user provided buffer and size
/// @param data pointer to user provided buffer
/// @param size size of the user provided buffer
/// @param ffn optional callback, called when the message is transfered (and can be deleted)
/// @param hint optional helper pointer that can be used in the callback
/// @param ffn callback, called when the message is transfered (and can be deleted)
/// @param obj optional helper pointer that can be used in the callback
/// @return pointer to FairMQMessage
inline FairMQMessagePtr NewMessage(void* data, int size, fairmq_free_fn* ffn, void* hint = nullptr) const
inline FairMQMessagePtr NewMessage(void* data, int size, fairmq_free_fn* ffn, void* obj = nullptr) const
{
return fTransportFactory->CreateMessage(data, size, ffn, obj);
}
template<typename... Args>
inline FairMQMessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args) const
{
return fChannels.at(channel).at(index).fTransportFactory->CreateMessage(std::forward<Args>(args)...);
}
static void FairMQNoCleanup(void* /*data*/, void* /*obj*/)
{
return fTransportFactory->CreateMessage(data, size, ffn, hint);
}
template<typename T>
@ -231,6 +253,23 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
return fTransportFactory->CreateMessage(const_cast<char*>(str.c_str()), str.length(), FairMQNoCleanup, nullptr);
}
template<typename T>
inline FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data) const
{
return fChannels.at(channel).at(index).fTransportFactory->CreateMessage(data, sizeof(T), FairMQNoCleanup, nullptr);
}
inline FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const std::string& str) const
{
return fChannels.at(channel).at(index).fTransportFactory->CreateMessage(const_cast<char*>(str.c_str()), str.length(), FairMQNoCleanup, nullptr);
}
template<typename T>
static void FairMQSimpleMsgCleanup(void* /*data*/, void* obj)
{
delete static_cast<T*>(obj);
}
template<typename T>
inline FairMQMessagePtr NewSimpleMessage(const T& data) const
{
@ -295,8 +334,11 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
/// Configures the device with a transport factory (DEPRECATED)
/// @param factory Pointer to the transport factory object
void SetTransport(FairMQTransportFactory* factory);
/// Configures the device with a transport factory
/// @param transport Transport string ("zeromq"/"nanomsg")
/// Adds a transport to the device if it doesn't exist
/// @param transport Transport string ("zeromq"/"nanomsg"/"shmem")
std::shared_ptr<FairMQTransportFactory> AddTransport(const std::string& transport);
/// Sets the default transport for the device
/// @param transport Transport string ("zeromq"/"nanomsg"/"shmem")
void SetTransport(const std::string& transport = "zeromq");
void SetConfig(FairMQProgOptions& config);
@ -317,6 +359,11 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
{
return (static_cast<T*>(this)->*memberFunction)(msg, index);
}));
if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
{
fInputChannelKeys.push_back(channelName);
}
}
void OnData(const std::string& channelName, InputMsgCallback);
@ -329,6 +376,11 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
{
return (static_cast<T*>(this)->*memberFunction)(parts, index);
}));
if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
{
fInputChannelKeys.push_back(channelName);
}
}
void OnData(const std::string& channelName, InputMultipartCallback);
@ -338,6 +390,7 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
protected:
std::string fId; ///< Device ID
std::string fNetworkInterface; ///< Network interface to use for dynamic binding
std::string fDefaultTransport; ///< Default transport for the device
int fMaxInitializationAttempts; ///< Timeout for the initialization
@ -348,9 +401,9 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
int fLogIntervalInMs; ///< Interval for logging the socket transfer rates
FairMQSocketPtr fCmdSocket; ///< Socket used for the internal unblocking mechanism
std::shared_ptr<FairMQTransportFactory> fTransportFactory; ///< Transport factory
std::unordered_map<FairMQ::Transport, std::shared_ptr<FairMQTransportFactory>> fTransports; ///< Container for transports
std::unordered_map<FairMQ::Transport, FairMQSocketPtr> fDeviceCmdSockets; ///< Sockets used for the internal unblocking mechanism
/// Additional user initialization (can be overloaded in child classes). Prefer to use InitTask().
virtual void Init();
@ -403,14 +456,8 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
/// Unblocks blocking channel send/receive calls
void Unblock();
/// Binds channel in the list
void BindChannels(std::list<FairMQChannel*>& chans);
/// Connects channel in the list
void ConnectChannels(std::list<FairMQChannel*>& chans);
/// Binds a single channel (used in InitWrapper)
bool BindChannel(FairMQChannel& ch);
/// Connects a single channel (used in InitWrapper)
bool ConnectChannel(FairMQChannel& ch);
/// Attach (bind/connect) channels in the list
void AttachChannels(std::list<FairMQChannel*>& chans);
/// Sets up and connects/binds a socket to an endpoint
/// return a string with the actual endpoint if it happens
@ -422,6 +469,14 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
/// to override default: prepend "@" to bind, "+" or ">" to connect endpoint.
bool AttachChannel(FairMQChannel& ch);
void HandleSingleChannelInput();
void HandleMultipleChannelInput();
void HandleMultipleTransportInput();
void PollForTransport(const FairMQTransportFactory* factory, const std::vector<std::string>& channelKeys);
bool HandleMsgInput(const std::string& chName, const InputMsgCallback& callback, int i) const;
bool HandleMultipartInput(const std::string& chName, const InputMultipartCallback& callback, int i) const;
/// Signal handler
void SignalHandler(int signal);
bool fCatchingSignals;
@ -432,6 +487,10 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
bool fDataCallbacks;
std::unordered_map<std::string, InputMsgCallback> fMsgInputs;
std::unordered_map<std::string, InputMultipartCallback> fMultipartInputs;
std::unordered_map<FairMQ::Transport, std::vector<std::string>> fMultitransportInputs;
std::vector<std::string> fInputChannelKeys;
std::mutex fMultitransportMutex;
std::atomic<bool> fMultitransportProceed;
};
#endif /* FAIRMQDEVICE_H_ */

4
fairmq/FairMQMessage.h
View File

@ -18,6 +18,8 @@
#include <cstddef> // for size_t
#include <memory> // unique_ptr
#include "FairMQTransports.h"
using fairmq_free_fn = void(void* data, void* hint);
class FairMQMessage
@ -34,6 +36,8 @@ class FairMQMessage
virtual void SetDeviceId(const std::string& deviceId) = 0;
virtual FairMQ::Transport GetType() const = 0;
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg) = 0;
virtual ~FairMQMessage() {};

8
fairmq/FairMQSocket.h
View File

@ -40,12 +40,10 @@ class FairMQSocket
virtual void Connect(const std::string& address) = 0;
virtual bool Attach(const std::string& address, bool serverish = false);
virtual int Send(FairMQMessage* msg, const std::string& flag = "") = 0;
virtual int Send(FairMQMessage* msg, const int flags = 0) = 0;
virtual int64_t Send(const std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0) = 0;
virtual int Send(FairMQMessagePtr& msg, const int flags = 0) = 0;
virtual int Receive(FairMQMessagePtr& msg, const int flags = 0) = 0;
virtual int Receive(FairMQMessage* msg, const std::string& flag = "") = 0;
virtual int Receive(FairMQMessage* msg, const int flags = 0) = 0;
virtual int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0) = 0;
virtual int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0) = 0;
virtual void* GetSocket() const = 0;

2
fairmq/FairMQStateMachine.h
View File

@ -99,7 +99,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
}
template <class Event, class FSM>
void on_exit(Event const&, FSM& fsm)
void on_exit(Event const&, FSM& /*fsm*/)
{
LOG(STATE) << "Exiting FairMQ state machine";
}

4
fairmq/FairMQTransportFactory.h
View File

@ -21,10 +21,10 @@
#include <unordered_map>
#include "FairMQMessage.h"
#include "FairMQChannel.h"
#include "FairMQSocket.h"
#include "FairMQPoller.h"
#include "FairMQLogger.h"
#include "FairMQTransports.h"
class FairMQChannel;
@ -41,6 +41,8 @@ class FairMQTransportFactory
virtual FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const = 0;
virtual FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const = 0;
virtual FairMQ::Transport GetType() const = 0;
virtual ~FairMQTransportFactory() {};
};

47
fairmq/FairMQTransports.h Normal file
View File

@ -0,0 +1,47 @@
/********************************************************************************
* 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 FAIRMQTRANSPORTS_H_
#define FAIRMQTRANSPORTS_H_
#include <string>
#include <unordered_map>
namespace FairMQ
{
enum class Transport
{
DEFAULT,
ZMQ,
NN,
SHM
};
static std::unordered_map<std::string, Transport> TransportTypes {
{ "default", Transport::DEFAULT },
{ "zeromq", Transport::ZMQ },
{ "nanomsg", Transport::NN },
{ "shmem", Transport::SHM }
};
}
namespace std
{
template <>
struct hash<FairMQ::Transport>
{
size_t operator()(const FairMQ::Transport& v) const
{
return hash<int>()(static_cast<int>(v));
}
};
}
#endif /* FAIRMQTRANSPORTS_H_ */

10
fairmq/devices/FairMQBenchmarkSampler.cxx
View File

@ -51,10 +51,10 @@ void FairMQBenchmarkSampler::Run()
uint64_t numSentMsgs = 0;
FairMQMessagePtr baseMsg(fTransportFactory->CreateMessage(fMsgSize));
// store the channel reference to avoid traversing the map on every loop iteration
const FairMQChannel& dataOutChannel = fChannels.at(fOutChannelName).at(0);
FairMQChannel& dataOutChannel = fChannels.at(fOutChannelName).at(0);
FairMQMessagePtr baseMsg(dataOutChannel.Transport()->CreateMessage(fMsgSize));
LOG(INFO) << "Starting the benchmark with message size of " << fMsgSize << " and number of messages " << fNumMsgs << ".";
auto tStart = chrono::high_resolution_clock::now();
@ -63,7 +63,7 @@ void FairMQBenchmarkSampler::Run()
{
if (fSameMessage)
{
FairMQMessagePtr msg(fTransportFactory->CreateMessage());
FairMQMessagePtr msg(dataOutChannel.Transport()->CreateMessage());
msg->Copy(baseMsg);
if (dataOutChannel.Send(msg) >= 0)
@ -80,7 +80,7 @@ void FairMQBenchmarkSampler::Run()
}
else
{
FairMQMessagePtr msg(fTransportFactory->CreateMessage(fMsgSize));
FairMQMessagePtr msg(dataOutChannel.Transport()->CreateMessage(fMsgSize));
if (dataOutChannel.Send(msg) >= 0)
{

4
fairmq/devices/FairMQSink.cxx
View File

@ -36,14 +36,14 @@ void FairMQSink::Run()
{
uint64_t numReceivedMsgs = 0;
// store the channel reference to avoid traversing the map on every loop iteration
const FairMQChannel& dataInChannel = fChannels.at(fInChannelName).at(0);
FairMQChannel& dataInChannel = fChannels.at(fInChannelName).at(0);
LOG(INFO) << "Starting the benchmark and expecting to receive " << fNumMsgs << " messages.";
auto tStart = chrono::high_resolution_clock::now();
while (CheckCurrentState(RUNNING))
{
FairMQMessagePtr msg(fTransportFactory->CreateMessage());
FairMQMessagePtr msg(dataInChannel.Transport()->CreateMessage());
if (dataInChannel.Receive(msg) >= 0)
{

7
fairmq/nanomsg/FairMQMessageNN.cxx
View File

@ -22,6 +22,8 @@
using namespace std;
static FairMQ::Transport gTransportType = FairMQ::Transport::NN;
string FairMQMessageNN::fDeviceID = string();
FairMQMessageNN::FairMQMessageNN()
@ -145,6 +147,11 @@ void FairMQMessageNN::SetDeviceId(const string& deviceId)
fDeviceID = deviceId;
}
FairMQ::Transport FairMQMessageNN::GetType() const
{
return gTransportType;
}
void FairMQMessageNN::Copy(const unique_ptr<FairMQMessage>& msg)
{
if (fMessage)

2
fairmq/nanomsg/FairMQMessageNN.h
View File

@ -41,6 +41,8 @@ class FairMQMessageNN : public FairMQMessage
virtual void SetDeviceId(const std::string& deviceId);
virtual FairMQ::Transport GetType() const;
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg);
virtual ~FairMQMessageNN();

23
fairmq/nanomsg/FairMQPollerNN.cxx
View File

@ -53,7 +53,7 @@ FairMQPollerNN::FairMQPollerNN(const vector<FairMQChannel>& channels)
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "nanomsg: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -104,7 +104,7 @@ FairMQPollerNN::FairMQPollerNN(const unordered_map<string, vector<FairMQChannel>
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "nanomsg: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -112,8 +112,8 @@ FairMQPollerNN::FairMQPollerNN(const unordered_map<string, vector<FairMQChannel>
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "At least one of the provided channel keys for poller initialization is invalid";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "nanomsg: at least one of the provided channel keys for poller initialization is invalid";
LOG(ERROR) << "nanomsg: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}
@ -150,7 +150,7 @@ FairMQPollerNN::FairMQPollerNN(const FairMQSocket& cmdSocket, const FairMQSocket
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "nanomsg: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -161,11 +161,12 @@ void FairMQPollerNN::Poll(const int timeout)
{
if (errno == ETERM)
{
LOG(DEBUG) << "polling exited, reason: " << nn_strerror(errno);
LOG(DEBUG) << "nanomsg: polling exited, reason: " << nn_strerror(errno);
}
else
{
LOG(ERROR) << "polling failed, reason: " << nn_strerror(errno);
LOG(ERROR) << "nanomsg: polling failed, reason: " << nn_strerror(errno);
throw std::runtime_error("nanomsg: polling failed");
}
}
}
@ -203,8 +204,8 @@ bool FairMQPollerNN::CheckInput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "nanomsg: invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "nanomsg: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}
@ -222,8 +223,8 @@ bool FairMQPollerNN::CheckOutput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "nanomsg: invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "nanomsg: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}

303
fairmq/nanomsg/FairMQSocketNN.cxx
View File

@ -29,6 +29,8 @@
using namespace std;
atomic<bool> FairMQSocketNN::fInterrupted(false);
FairMQSocketNN::FairMQSocketNN(const string& type, const string& name, const int numIoThreads, const string& id /*= ""*/)
: FairMQSocket(0, 0, NN_DONTWAIT)
, fSocket(-1)
@ -71,6 +73,18 @@ FairMQSocketNN::FairMQSocketNN(const string& type, const string& name, const int
}
}
int sndTimeout = 700;
if (nn_setsockopt(fSocket, NN_SOL_SOCKET, NN_SNDTIMEO, &sndTimeout, sizeof(sndTimeout)) != 0)
{
LOG(ERROR) << "Failed setting NN_SNDTIMEO socket option, reason: " << nn_strerror(errno);
}
int rcvTimeout = 700;
if (nn_setsockopt(fSocket, NN_SOL_SOCKET, NN_RCVTIMEO, &rcvTimeout, sizeof(rcvTimeout)) != 0)
{
LOG(ERROR) << "Failed setting NN_RCVTIMEO socket option, reason: " << nn_strerror(errno);
}
#ifdef NN_RCVMAXSIZE
int rcvSize = -1;
nn_setsockopt(fSocket, NN_SOL_SOCKET, NN_RCVMAXSIZE, &rcvSize, sizeof(rcvSize));
@ -108,105 +122,155 @@ void FairMQSocketNN::Connect(const string& address)
}
}
int FairMQSocketNN::Send(FairMQMessage* msg, const string& flag)
int FairMQSocketNN::Send(FairMQMessagePtr& msg, const int flags)
{
return Send(msg, GetConstant(flag));
int nbytes = -1;
while (true)
{
void* ptr = msg->GetMessage();
nbytes = nn_send(fSocket, &ptr, NN_MSG, flags);
if (nbytes >= 0)
{
fBytesTx += nbytes;
++fMessagesTx;
static_cast<FairMQMessageNN*>(msg.get())->fReceiving = false;
return nbytes;
}
else if (nn_errno() == ETIMEDOUT)
{
if (!fInterrupted && ((flags & NN_DONTWAIT) == 0))
{
continue;
}
else
{
return -2;
}
}
else if (nn_errno() == EAGAIN)
{
return -2;
}
else if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
}
}
}
int FairMQSocketNN::Send(FairMQMessage* msg, const int flags)
int FairMQSocketNN::Receive(FairMQMessagePtr& msg, const int flags)
{
void* ptr = msg->GetMessage();
int nbytes = nn_send(fSocket, &ptr, NN_MSG, flags);
if (nbytes >= 0)
int nbytes = -1;
while (true)
{
fBytesTx += nbytes;
++fMessagesTx;
static_cast<FairMQMessageNN*>(msg)->fReceiving = false;
return nbytes;
void* ptr = NULL;
nbytes = nn_recv(fSocket, &ptr, NN_MSG, flags);
if (nbytes >= 0)
{
fBytesRx += nbytes;
++fMessagesRx;
msg->SetMessage(ptr, nbytes);
static_cast<FairMQMessageNN*>(msg.get())->fReceiving = true;
return nbytes;
}
else if (nn_errno() == ETIMEDOUT)
{
if (!fInterrupted && ((flags & NN_DONTWAIT) == 0))
{
continue;
}
else
{
return -2;
}
}
else if (nn_errno() == EAGAIN)
{
return -2;
}
else if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
}
}
if (nn_errno() == EAGAIN)
{
return -2;
}
if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
}
int64_t FairMQSocketNN::Send(const vector<unique_ptr<FairMQMessage>>& msgVec, const int flags)
int64_t FairMQSocketNN::Send(vector<unique_ptr<FairMQMessage>>& msgVec, const int flags)
{
#ifdef MSGPACK_FOUND
const unsigned int vecSize = msgVec.size();
// create msgpack simple buffer
msgpack::sbuffer sbuf;
// create msgpack packer
msgpack::packer<msgpack::sbuffer> packer(&sbuf);
// pack all parts into a single msgpack simple buffer
for (unsigned int i = 0; i < msgVec.size(); ++i)
for (unsigned int i = 0; i < vecSize; ++i)
{
static_cast<FairMQMessageNN*>(msgVec[i].get())->fReceiving = false;
packer.pack_bin(msgVec[i]->GetSize());
packer.pack_bin_body(static_cast<char*>(msgVec[i]->GetData()), msgVec[i]->GetSize());
}
int64_t nbytes = nn_send(fSocket, sbuf.data(), sbuf.size(), flags);
if (nbytes >= 0)
int64_t nbytes = -1;
while (true)
{
fBytesTx += nbytes;
++fMessagesTx;
return nbytes;
nbytes = nn_send(fSocket, sbuf.data(), sbuf.size(), flags);
if (nbytes >= 0)
{
fBytesTx += nbytes;
++fMessagesTx;
return nbytes;
}
else if (nn_errno() == ETIMEDOUT)
{
if (!fInterrupted && ((flags & NN_DONTWAIT) == 0))
{
continue;
}
else
{
return -2;
}
}
else if (nn_errno() == EAGAIN)
{
return -2;
}
else if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
}
}
if (nn_errno() == EAGAIN)
{
return -2;
}
if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
#else /*MSGPACK_FOUND*/
LOG(ERROR) << "Cannot send message from vector of size " << msgVec.size() << " and flags " << flags << " with nanomsg multipart because MessagePack is not available.";
LOG(ERROR) << "Cannot send message from vector of size " << vecSize << " and flags " << flags << " with nanomsg multipart because MessagePack is not available.";
exit(EXIT_FAILURE);
#endif /*MSGPACK_FOUND*/
}
int FairMQSocketNN::Receive(FairMQMessage* msg, const string& flag)
{
return Receive(msg, GetConstant(flag));
}
int FairMQSocketNN::Receive(FairMQMessage* msg, const int flags)
{
void* ptr = NULL;
int nbytes = nn_recv(fSocket, &ptr, NN_MSG, flags);
if (nbytes >= 0)
{
fBytesRx += nbytes;
++fMessagesRx;
msg->SetMessage(ptr, nbytes);
static_cast<FairMQMessageNN*>(msg)->fReceiving = true;
return nbytes;
}
if (nn_errno() == EAGAIN)
{
return -2;
}
if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
}
int64_t FairMQSocketNN::Receive(vector<unique_ptr<FairMQMessage>>& msgVec, const int flags)
{
#ifdef MSGPACK_FOUND
@ -217,51 +281,68 @@ int64_t FairMQSocketNN::Receive(vector<unique_ptr<FairMQMessage>>& msgVec, const
msgVec.clear();
}
// pointer to point to received message buffer
char* ptr = NULL;
// receive the message into a buffer allocated by nanomsg and let ptr point to it
int nbytes = nn_recv(fSocket, &ptr, NN_MSG, flags);
if (nbytes >= 0) // if no errors or non-blocking timeouts
while (true)
{
// store statistics on how many bytes received
fBytesRx += nbytes;
// store statistics on how many messages received (count messages instead of parts)
++fMessagesRx;
// offset to be used by msgpack to handle separate chunks
size_t offset = 0;
while (offset != static_cast<size_t>(nbytes)) // continue until all parts have been read
// pointer to point to received message buffer
char* ptr = NULL;
// receive the message into a buffer allocated by nanomsg and let ptr point to it
int nbytes = nn_recv(fSocket, &ptr, NN_MSG, flags);
if (nbytes >= 0) // if no errors or non-blocking timeouts
{
// vector of chars to hold blob (unlike char*/void* this type can be converted to by msgpack)
std::vector<char> buf;
// store statistics on how many bytes received
fBytesRx += nbytes;
// store statistics on how many messages received (count messages instead of parts)
++fMessagesRx;
// unpack and convert chunk
msgpack::unpacked result;
unpack(result, ptr, nbytes, offset);
msgpack::object object(result.get());
object.convert(buf);
// get the single message size
size_t size = buf.size() * sizeof(char);
unique_ptr<FairMQMessage> part(new FairMQMessageNN(size));
static_cast<FairMQMessageNN*>(part.get())->fReceiving = true;
memcpy(part->GetData(), buf.data(), size);
msgVec.push_back(move(part));
// offset to be used by msgpack to handle separate chunks
size_t offset = 0;
while (offset != static_cast<size_t>(nbytes)) // continue until all parts have been read
{
// vector of chars to hold blob (unlike char*/void* this type can be converted to by msgpack)
std::vector<char> buf;
// unpack and convert chunk
msgpack::unpacked result;
unpack(result, ptr, nbytes, offset);
msgpack::object object(result.get());
object.convert(buf);
// get the single message size
size_t size = buf.size() * sizeof(char);
unique_ptr<FairMQMessage> part(new FairMQMessageNN(size));
static_cast<FairMQMessageNN*>(part.get())->fReceiving = true;
memcpy(part->GetData(), buf.data(), size);
msgVec.push_back(move(part));
}
nn_freemsg(ptr);
return nbytes;
}
else if (nn_errno() == ETIMEDOUT)
{
if (!fInterrupted && ((flags & NN_DONTWAIT) == 0))
{
continue;
}
else
{
return -2;
}
}
else if (nn_errno() == EAGAIN)
{
return -2;
}
else if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
}
nn_freemsg(ptr);
return nbytes;
}
if (nn_errno() == EAGAIN)
{
return -2;
}
if (nn_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << nn_strerror(errno);
return nbytes;
#else /*MSGPACK_FOUND*/
LOG(ERROR) << "Cannot receive message into vector of size " << msgVec.size() << " and flags " << flags << " with nanomsg multipart because MessagePack is not available.";
exit(EXIT_FAILURE);
@ -280,10 +361,12 @@ void FairMQSocketNN::Terminate()
void FairMQSocketNN::Interrupt()
{
fInterrupted = true;
}
void FairMQSocketNN::Resume()
{
fInterrupted = false;
}
void* FairMQSocketNN::GetSocket() const

10
fairmq/nanomsg/FairMQSocketNN.h
View File

@ -19,6 +19,7 @@
#include <atomic>
#include "FairMQSocket.h"
#include "FairMQMessage.h"
class FairMQSocketNN : public FairMQSocket
{
@ -32,12 +33,10 @@ class FairMQSocketNN : public FairMQSocket
virtual bool Bind(const std::string& address);
virtual void Connect(const std::string& address);
virtual int Send(FairMQMessage* msg, const std::string& flag = "");
virtual int Send(FairMQMessage* msg, const int flags = 0);
virtual int64_t Send(const std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual int Send(FairMQMessagePtr& msg, const int flags = 0);
virtual int Receive(FairMQMessagePtr& msg, const int flags = 0);
virtual int Receive(FairMQMessage* msg, const std::string& flag = "");
virtual int Receive(FairMQMessage* msg, const int flags = 0);
virtual int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual void* GetSocket() const;
@ -72,6 +71,7 @@ class FairMQSocketNN : public FairMQSocket
std::atomic<unsigned long> fBytesRx;
std::atomic<unsigned long> fMessagesTx;
std::atomic<unsigned long> fMessagesRx;
static std::atomic<bool> fInterrupted;
};
#endif /* FAIRMQSOCKETNN_H_ */

9
fairmq/nanomsg/FairMQTransportFactoryNN.cxx
View File

@ -16,9 +16,11 @@
using namespace std;
static FairMQ::Transport gTransportType = FairMQ::Transport::NN;
FairMQTransportFactoryNN::FairMQTransportFactoryNN()
{
LOG(INFO) << "Using nanomsg library";
LOG(DEBUG) << "Transport: Using nanomsg library";
}
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage() const
@ -55,3 +57,8 @@ FairMQPollerPtr FairMQTransportFactoryNN::CreatePoller(const FairMQSocket& cmdSo
{
return unique_ptr<FairMQPoller>(new FairMQPollerNN(cmdSocket, dataSocket));
}
FairMQ::Transport FairMQTransportFactoryNN::GetType() const
{
return gTransportType;
}

3
fairmq/nanomsg/FairMQTransportFactoryNN.h
View File

@ -16,6 +16,7 @@
#define FAIRMQTRANSPORTFACTORYNN_H_
#include <vector>
#include <string>
#include "FairMQTransportFactory.h"
#include "FairMQMessageNN.h"
@ -37,6 +38,8 @@ class FairMQTransportFactoryNN : public FairMQTransportFactory
virtual FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const;
virtual FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const;
virtual FairMQ::Transport GetType() const;
virtual ~FairMQTransportFactoryNN() {};
};

9
fairmq/options/FairMQParser.cxx
View File

@ -228,6 +228,7 @@ void ChannelParser(const boost::property_tree::ptree& tree, FairMQMap& channelMa
commonChannel.UpdateType(q.second.get<string>("type", commonChannel.GetType()));
commonChannel.UpdateMethod(q.second.get<string>("method", commonChannel.GetMethod()));
commonChannel.UpdateAddress(q.second.get<string>("address", commonChannel.GetAddress()));
commonChannel.UpdateTransport(q.second.get<string>("transport", commonChannel.GetTransport()));
commonChannel.UpdateSndBufSize(q.second.get<int>("sndBufSize", commonChannel.GetSndBufSize()));
commonChannel.UpdateRcvBufSize(q.second.get<int>("rcvBufSize", commonChannel.GetRcvBufSize()));
commonChannel.UpdateSndKernelSize(q.second.get<int>("sndKernelSize", commonChannel.GetSndKernelSize()));
@ -246,6 +247,7 @@ void ChannelParser(const boost::property_tree::ptree& tree, FairMQMap& channelMa
LOG(DEBUG) << "\ttype = " << commonChannel.GetType();
LOG(DEBUG) << "\tmethod = " << commonChannel.GetMethod();
LOG(DEBUG) << "\taddress = " << commonChannel.GetAddress();
LOG(DEBUG) << "\ttransport = " << commonChannel.GetTransport();
LOG(DEBUG) << "\tsndBufSize = " << commonChannel.GetSndBufSize();
LOG(DEBUG) << "\trcvBufSize = " << commonChannel.GetRcvBufSize();
LOG(DEBUG) << "\tsndKernelSize = " << commonChannel.GetSndKernelSize();
@ -289,6 +291,7 @@ void ChannelParser(const boost::property_tree::ptree& tree, FairMQMap& channelMa
commonChannel.UpdateType(p.second.get<string>("type", commonChannel.GetType()));
commonChannel.UpdateMethod(p.second.get<string>("method", commonChannel.GetMethod()));
commonChannel.UpdateAddress(p.second.get<string>("address", commonChannel.GetAddress()));
commonChannel.UpdateTransport(p.second.get<string>("transport", commonChannel.GetTransport()));
commonChannel.UpdateSndBufSize(p.second.get<int>("sndBufSize", commonChannel.GetSndBufSize()));
commonChannel.UpdateRcvBufSize(p.second.get<int>("rcvBufSize", commonChannel.GetRcvBufSize()));
commonChannel.UpdateSndKernelSize(p.second.get<int>("sndKernelSize", commonChannel.GetSndKernelSize()));
@ -308,6 +311,7 @@ void ChannelParser(const boost::property_tree::ptree& tree, FairMQMap& channelMa
LOG(DEBUG) << "\ttype = " << commonChannel.GetType();
LOG(DEBUG) << "\tmethod = " << commonChannel.GetMethod();
LOG(DEBUG) << "\taddress = " << commonChannel.GetAddress();
LOG(DEBUG) << "\ttransport = " << commonChannel.GetTransport();
LOG(DEBUG) << "\tsndBufSize = " << commonChannel.GetSndBufSize();
LOG(DEBUG) << "\trcvBufSize = " << commonChannel.GetRcvBufSize();
LOG(DEBUG) << "\tsndKernelSize = " << commonChannel.GetSndKernelSize();
@ -348,6 +352,7 @@ void SocketParser(const boost::property_tree::ptree& tree, vector<FairMQChannel>
channel.UpdateType(q.second.get<string>("type", channel.GetType()));
channel.UpdateMethod(q.second.get<string>("method", channel.GetMethod()));
channel.UpdateAddress(q.second.get<string>("address", channel.GetAddress()));
channel.UpdateTransport(q.second.get<string>("transport", channel.GetTransport()));
channel.UpdateSndBufSize(q.second.get<int>("sndBufSize", channel.GetSndBufSize()));
channel.UpdateRcvBufSize(q.second.get<int>("rcvBufSize", channel.GetRcvBufSize()));
channel.UpdateSndKernelSize(q.second.get<int>("sndKernelSize", channel.GetSndKernelSize()));
@ -358,6 +363,7 @@ void SocketParser(const boost::property_tree::ptree& tree, vector<FairMQChannel>
LOG(DEBUG) << "\ttype = " << channel.GetType();
LOG(DEBUG) << "\tmethod = " << channel.GetMethod();
LOG(DEBUG) << "\taddress = " << channel.GetAddress();
LOG(DEBUG) << "\ttransport = " << channel.GetTransport();
LOG(DEBUG) << "\tsndBufSize = " << channel.GetSndBufSize();
LOG(DEBUG) << "\trcvBufSize = " << channel.GetRcvBufSize();
LOG(DEBUG) << "\tsndKernelSize = " << channel.GetSndKernelSize();
@ -378,6 +384,7 @@ void SocketParser(const boost::property_tree::ptree& tree, vector<FairMQChannel>
channel.UpdateType(p.second.get<string>("type", channel.GetType()));
channel.UpdateMethod(p.second.get<string>("method", channel.GetMethod()));
channel.UpdateAddress(p.second.get<string>("address", channel.GetAddress()));
channel.UpdateTransport(p.second.get<string>("transport", channel.GetTransport()));
channel.UpdateSndBufSize(p.second.get<int>("sndBufSize", channel.GetSndBufSize()));
channel.UpdateRcvBufSize(p.second.get<int>("rcvBufSize", channel.GetRcvBufSize()));
channel.UpdateSndKernelSize(p.second.get<int>("sndKernelSize", channel.GetSndKernelSize()));
@ -388,6 +395,7 @@ void SocketParser(const boost::property_tree::ptree& tree, vector<FairMQChannel>
LOG(DEBUG) << "\ttype = " << channel.GetType();
LOG(DEBUG) << "\tmethod = " << channel.GetMethod();
LOG(DEBUG) << "\taddress = " << channel.GetAddress();
LOG(DEBUG) << "\ttransport = " << channel.GetTransport();
LOG(DEBUG) << "\tsndBufSize = " << channel.GetSndBufSize();
LOG(DEBUG) << "\trcvBufSize = " << channel.GetRcvBufSize();
LOG(DEBUG) << "\tsndKernelSize = " << channel.GetSndKernelSize();
@ -414,6 +422,7 @@ void SocketParser(const boost::property_tree::ptree& tree, vector<FairMQChannel>
LOG(DEBUG) << "\ttype = " << channel.GetType();
LOG(DEBUG) << "\tmethod = " << channel.GetMethod();
LOG(DEBUG) << "\taddress = " << channel.GetAddress();
LOG(DEBUG) << "\ttransport = " << channel.GetTransport();
LOG(DEBUG) << "\tsndBufSize = " << channel.GetSndBufSize();
LOG(DEBUG) << "\trcvBufSize = " << channel.GetRcvBufSize();
LOG(DEBUG) << "\tsndKernelSize = " << channel.GetSndKernelSize();

9
fairmq/options/FairMQProgOptions.cxx
View File

@ -233,6 +233,7 @@ void FairMQProgOptions::UpdateMQValues()
string typeKey = p.first + "." + to_string(index) + ".type";
string methodKey = p.first + "." + to_string(index) + ".method";
string addressKey = p.first + "." + to_string(index) + ".address";
string transportKey = p.first + "." + to_string(index) + ".transport";
string sndBufSizeKey = p.first + "." + to_string(index) + ".sndBufSize";
string rcvBufSizeKey = p.first + "." + to_string(index) + ".rcvBufSize";
string sndKernelSizeKey = p.first + "." + to_string(index) + ".sndKernelSize";
@ -242,6 +243,7 @@ void FairMQProgOptions::UpdateMQValues()
fMQKeyMap[typeKey] = make_tuple(p.first, index, "type");
fMQKeyMap[methodKey] = make_tuple(p.first, index, "method");
fMQKeyMap[addressKey] = make_tuple(p.first, index, "address");
fMQKeyMap[transportKey] = make_tuple(p.first, index, "transport");
fMQKeyMap[sndBufSizeKey] = make_tuple(p.first, index, "sndBufSize");
fMQKeyMap[rcvBufSizeKey] = make_tuple(p.first, index, "rcvBufSize");
fMQKeyMap[sndKernelSizeKey] = make_tuple(p.first, index, "sndKernelSize");
@ -251,6 +253,7 @@ void FairMQProgOptions::UpdateMQValues()
UpdateVarMap<string>(typeKey, channel.GetType());
UpdateVarMap<string>(methodKey, channel.GetMethod());
UpdateVarMap<string>(addressKey, channel.GetAddress());
UpdateVarMap<string>(transportKey, channel.GetTransport());
//UpdateVarMap<string>(sndBufSizeKey, to_string(channel.GetSndBufSize()));// string API
@ -384,6 +387,12 @@ int FairMQProgOptions::UpdateChannelMap(const string& channelName, int index, co
fFairMQMap.at(channelName).at(index).UpdateAddress(val);
return 0;
}
if (member == "transport")
{
fFairMQMap.at(channelName).at(index).UpdateTransport(val);
return 0;
}
else
{
//if we get there it means something is wrong

6
fairmq/plugins/control/FairMQDDSControlPlugin.cxx
View File

@ -35,8 +35,8 @@ class FairMQControlPluginDDS
return fInstance;
}
static void ResetInstance()
{
static void ResetInstance()
{
try
{
delete fInstance;
@ -47,7 +47,7 @@ class FairMQControlPluginDDS
LOG(ERROR) << "Error: " << e.what() << endl;
return;
}
}
}
void Init(FairMQDevice& device)
{

41
fairmq/run/startMQBenchmark.sh.in
View File

@ -4,6 +4,10 @@ numMsgs="0"
msgSize="1000000"
transport="zeromq"
sameMsg="true"
affinity="false"
affinitySamp=""
affinitySink=""
if [[ $1 =~ ^[0-9]+$ ]]; then
msgSize=$1
@ -21,17 +25,40 @@ if [[ $4 =~ ^[a-z]+$ ]]; then
sameMsg=$4
fi
echo "Starting benchmark with message size of $msgSize bytes ($numMsgs messages) and $transport transport."
echo "Using $transport transport."
if [[ $5 =~ ^[a-z]+$ ]]; then
affinity=$5
fi
echo "Starting benchmark with following settings:"
echo ""
echo "message size: $msgSize bytes"
if [ $numMsgs = 0 ]; then
echo "Unlimited number of messages."
echo "number of messages: unlimited"
else
echo "Number of messages: $numMsgs."
echo "number of messages: $numMsgs"
fi
echo "transport: $transport"
if [ $sameMsg = "true" ]; then
echo "resend same message: yes, using Copy() method to resend the same message"
else
echo "resend same message: no, allocating each message separately"
fi
if [ $affinity = "true" ]; then
affinitySamp="taskset -c 0"
affinitySink="taskset -c 1"
echo "affinity: assigning sampler to core 0, sink to core 1"
else
echo ""
fi
echo ""
echo "Usage: startBenchmark [message size=1000000] [number of messages=0] [transport=zeromq/nanomsg/shmem] [resend same message=true]"
echo "Usage: startBenchmark [message size=1000000] [number of messages=0] [transport=zeromq/nanomsg/shmem] [resend same message=true] [affinity=false]"
SAMPLER="bsampler"
SAMPLER+=" --id bsampler1"
@ -43,7 +70,7 @@ SAMPLER+=" --same-msg $sameMsg"
# SAMPLER+=" --msg-rate 1000"
SAMPLER+=" --num-msgs $numMsgs"
SAMPLER+=" --mq-config @CMAKE_BINARY_DIR@/bin/config/benchmark.json"
xterm -geometry 80x23+0+0 -hold -e @CMAKE_BINARY_DIR@/bin/$SAMPLER &
xterm -geometry 90x23+0+0 -hold -e $affinitySamp @CMAKE_BINARY_DIR@/bin/$SAMPLER &
SINK="sink"
SINK+=" --id sink1"
@ -52,4 +79,4 @@ SINK+=" --id sink1"
SINK+=" --transport $transport"
SINK+=" --num-msgs $numMsgs"
SINK+=" --mq-config @CMAKE_BINARY_DIR@/bin/config/benchmark.json"
xterm -geometry 80x23+500+0 -hold -e @CMAKE_BINARY_DIR@/bin/$SINK &
xterm -geometry 90x23+550+0 -hold -e $affinitySink @CMAKE_BINARY_DIR@/bin/$SINK &

6
fairmq/shmem/FairMQContextSHM.cxx
View File

@ -7,6 +7,8 @@
********************************************************************************/
#include <zmq.h>
#include <cstdio>
#include <boost/interprocess/managed_shared_memory.hpp>
#include "FairMQLogger.h"
@ -46,11 +48,11 @@ FairMQContextSHM::~FairMQContextSHM()
if (boost::interprocess::shared_memory_object::remove("FairMQSharedMemory"))
{
LOG(INFO) << "Successfully removed shared memory after the device has stopped.";
printf("Successfully removed shared memory after the device has stopped.\n");
}
else
{
LOG(INFO) << "Did not remove shared memory after the device stopped. Still in use?";
printf("Did not remove shared memory after the device stopped. Already removed?\n");
}
}

201
fairmq/shmem/FairMQMessageSHM.cxx
View File

@ -14,45 +14,60 @@
using namespace std;
using namespace FairMQ::shmem;
uint64_t FairMQMessageSHM::fMessageID = 0;
string FairMQMessageSHM::fDeviceID = string();
static FairMQ::Transport gTransportType = FairMQ::Transport::SHM;
// uint64_t FairMQMessageSHM::fMessageID = 0;
// string FairMQMessageSHM::fDeviceID = string();
atomic<bool> FairMQMessageSHM::fInterrupted(false);
FairMQMessageSHM::FairMQMessageSHM()
: fMessage()
, fOwner(nullptr)
, fReceiving(false)
// , fOwner(nullptr)
// , fReceiving(false)
, fQueued(false)
, fMetaCreated(false)
, fHandle()
, fChunkSize(0)
, fLocalPtr(nullptr)
{
if (zmq_msg_init(&fMessage) != 0)
{
LOG(ERROR) << "failed initializing message, reason: " << zmq_strerror(errno);
}
fMetaCreated = true;
}
void FairMQMessageSHM::StringDeleter(void* /*data*/, void* str)
{
delete static_cast<string*>(str);
}
// void FairMQMessageSHM::StringDeleter(void* /*data*/, void* str)
// {
// delete static_cast<string*>(str);
// }
FairMQMessageSHM::FairMQMessageSHM(const size_t size)
: fMessage()
, fOwner(nullptr)
, fReceiving(false)
// , fOwner(nullptr)
// , fReceiving(false)
, fQueued(false)
, fMetaCreated(false)
, fHandle()
, fChunkSize(0)
, fLocalPtr(nullptr)
{
InitializeChunk(size);
}
FairMQMessageSHM::FairMQMessageSHM(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
: fMessage()
, fOwner(nullptr)
, fReceiving(false)
// , fOwner(nullptr)
// , fReceiving(false)
, fQueued(false)
, fMetaCreated(false)
, fHandle()
, fChunkSize(0)
, fLocalPtr(nullptr)
{
if (InitializeChunk(size))
{
memcpy(fOwner->fPtr->GetData(), data, size);
memcpy(fLocalPtr, data, size);
if (ffn)
{
ffn(data, hint);
@ -66,66 +81,76 @@ FairMQMessageSHM::FairMQMessageSHM(void* data, const size_t size, fairmq_free_fn
bool FairMQMessageSHM::InitializeChunk(const size_t size)
{
string chunkID = fDeviceID + "c" + to_string(fMessageID);
string* ownerID = new string(fDeviceID + "o" + to_string(fMessageID));
// string chunkID = fDeviceID + "c" + to_string(fMessageID);
// string* ownerID = new string(fDeviceID + "o" + to_string(fMessageID));
bool success = false;
while (!success)
while (!fHandle)
{
try
{
fOwner = Manager::Instance().Segment()->construct<ShPtrOwner>(ownerID->c_str())(
make_managed_shared_ptr(Manager::Instance().Segment()->construct<Chunk>(chunkID.c_str())(size),
*(Manager::Instance().Segment())));
success = true;
fLocalPtr = Manager::Instance().Segment()->allocate(size);
// fOwner = Manager::Instance().Segment()->construct<ShPtrOwner>(ownerID->c_str())(
// make_managed_shared_ptr(Manager::Instance().Segment()->construct<Chunk>(chunkID.c_str())(size),
// *(Manager::Instance().Segment())));
}
catch (bipc::bad_alloc& ba)
{
LOG(WARN) << "Shared memory full...";
// LOG(WARN) << "Shared memory full...";
this_thread::sleep_for(chrono::milliseconds(50));
if (fInterrupted)
{
break;
return false;
}
else
{
continue;
}
}
fHandle = Manager::Instance().Segment()->get_handle_from_address(fLocalPtr);
}
if (success)
fChunkSize = size;
if (zmq_msg_init_size(&fMessage, sizeof(MetaHeader)) != 0)
{
if (zmq_msg_init_data(&fMessage, const_cast<char*>(ownerID->c_str()), ownerID->length(), StringDeleter, ownerID) != 0)
{
LOG(ERROR) << "failed initializing meta message, reason: " << zmq_strerror(errno);
}
++fMessageID;
LOG(ERROR) << "failed initializing meta message, reason: " << zmq_strerror(errno);
return false;
}
MetaHeader* metaPtr = new(zmq_msg_data(&fMessage)) MetaHeader();
metaPtr->fSize = size;
metaPtr->fHandle = fHandle;
return success;
// if (zmq_msg_init_data(&fMessage, const_cast<char*>(ownerID->c_str()), ownerID->length(), StringDeleter, ownerID) != 0)
// {
// LOG(ERROR) << "failed initializing meta message, reason: " << zmq_strerror(errno);
// }
fMetaCreated = true;
// ++fMessageID;
return true;
}
void FairMQMessageSHM::Rebuild()
{
CloseMessage();
fReceiving = false;
// fReceiving = false;
fQueued = false;
if (zmq_msg_init(&fMessage) != 0)
{
LOG(ERROR) << "failed initializing message, reason: " << zmq_strerror(errno);
}
fMetaCreated = true;
}
void FairMQMessageSHM::Rebuild(const size_t size)
{
CloseMessage();
fReceiving = false;
// fReceiving = false;
fQueued = false;
InitializeChunk(size);
@ -135,12 +160,12 @@ void FairMQMessageSHM::Rebuild(void* data, const size_t size, fairmq_free_fn* ff
{
CloseMessage();
fReceiving = false;
// fReceiving = false;
fQueued = false;
if (InitializeChunk(size))
{
memcpy(fOwner->fPtr->GetData(), data, size);
memcpy(fLocalPtr, data, size);
if (ffn)
{
ffn(data, hint);
@ -159,27 +184,42 @@ void* FairMQMessageSHM::GetMessage()
void* FairMQMessageSHM::GetData()
{
if (fOwner)
if (fLocalPtr)
{
return fOwner->fPtr->GetData();
return fLocalPtr;
}
else if (fHandle)
{
return Manager::Instance().Segment()->get_address_from_handle(fHandle);
}
else
{
LOG(ERROR) << "Trying to get data of an empty shared memory message";
exit(EXIT_FAILURE);
// LOG(ERROR) << "Trying to get data of an empty shared memory message";
return nullptr;
}
// if (fOwner)
// {
// return fOwner->fPtr->GetData();
// }
// else
// {
// LOG(ERROR) << "Trying to get data of an empty shared memory message";
// exit(EXIT_FAILURE);
// }
}
size_t FairMQMessageSHM::GetSize()
{
if (fOwner)
{
return fOwner->fPtr->GetSize();
}
else
{
return 0;
}
return fChunkSize;
// if (fOwner)
// {
// return fOwner->fPtr->GetSize();
// }
// else
// {
// return 0;
// }
}
void FairMQMessageSHM::SetMessage(void*, const size_t)
@ -187,21 +227,26 @@ void FairMQMessageSHM::SetMessage(void*, const size_t)
// dummy method to comply with the interface. functionality not allowed in zeromq.
}
void FairMQMessageSHM::SetDeviceId(const string& deviceId)
void FairMQMessageSHM::SetDeviceId(const string& /*deviceId*/)
{
fDeviceID = deviceId;
// fDeviceID = deviceId;
}
FairMQ::Transport FairMQMessageSHM::GetType() const
{
return gTransportType;
}
void FairMQMessageSHM::Copy(const unique_ptr<FairMQMessage>& msg)
{
if (!fOwner)
if (!fHandle)
{
FairMQ::shmem::ShPtrOwner* otherOwner = static_cast<FairMQMessageSHM*>(msg.get())->fOwner;
if (otherOwner)
bipc::managed_shared_memory::handle_t otherHandle = static_cast<FairMQMessageSHM*>(msg.get())->fHandle;
if (otherHandle)
{
if (InitializeChunk(otherOwner->fPtr->GetSize()))
if (InitializeChunk(msg->GetSize()))
{
memcpy(fOwner->fPtr->GetData(), otherOwner->fPtr->GetData(), otherOwner->fPtr->GetSize());
memcpy(GetData(), msg->GetData(), msg->GetSize());
}
}
else
@ -266,31 +311,35 @@ void FairMQMessageSHM::Copy(const unique_ptr<FairMQMessage>& msg)
void FairMQMessageSHM::CloseMessage()
{
if (fReceiving)
{
if (fOwner)
// if (fReceiving)
// {
// if (fOwner)
// {
// Manager::Instance().Segment()->destroy_ptr(fOwner);
// fOwner = nullptr;
// }
// else
// {
// LOG(ERROR) << "No shared pointer owner when closing a received message";
// }
// }
// else
// {
if (fHandle && !fQueued)
{
Manager::Instance().Segment()->destroy_ptr(fOwner);
fOwner = nullptr;
// LOG(WARN) << "Destroying unsent message";
// Manager::Instance().Segment()->destroy_ptr(fHandle);
Manager::Instance().Segment()->deallocate(Manager::Instance().Segment()->get_address_from_handle(fHandle));
fHandle = 0;
}
else
{
LOG(ERROR) << "No shared pointer owner when closing a received message";
}
}
else
{
if (fOwner && !fQueued)
{
LOG(WARN) << "Destroying unsent message";
Manager::Instance().Segment()->destroy_ptr(fOwner);
fOwner = nullptr;
}
}
// }
if (zmq_msg_close(&fMessage) != 0)
if (fMetaCreated)
{
LOG(ERROR) << "failed closing message, reason: " << zmq_strerror(errno);
if (zmq_msg_close(&fMessage) != 0)
{
LOG(ERROR) << "failed closing message, reason: " << zmq_strerror(errno);
}
}
}

16
fairmq/shmem/FairMQMessageSHM.h
View File

@ -42,22 +42,28 @@ class FairMQMessageSHM : public FairMQMessage
virtual void SetDeviceId(const std::string& deviceId);
virtual FairMQ::Transport GetType() const;
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg);
void CloseMessage();
virtual ~FairMQMessageSHM();
static void StringDeleter(void* data, void* str);
// static void StringDeleter(void* data, void* str);
private:
zmq_msg_t fMessage;
FairMQ::shmem::ShPtrOwner* fOwner;
static uint64_t fMessageID;
static std::string fDeviceID;
bool fReceiving;
// FairMQ::shmem::ShPtrOwner* fOwner;
// static uint64_t fMessageID;
// static std::string fDeviceID;
// bool fReceiving;
bool fQueued;
bool fMetaCreated;
static std::atomic<bool> fInterrupted;
bipc::managed_shared_memory::handle_t fHandle;
size_t fChunkSize;
void* fLocalPtr;
};
#endif /* FAIRMQMESSAGESHM_H_ */

23
fairmq/shmem/FairMQPollerSHM.cxx
View File

@ -51,7 +51,7 @@ FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel>& channels)
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "shmem: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -105,7 +105,7 @@ FairMQPollerSHM::FairMQPollerSHM(const unordered_map<string, vector<FairMQChanne
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "shmem: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -113,8 +113,8 @@ FairMQPollerSHM::FairMQPollerSHM(const unordered_map<string, vector<FairMQChanne
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "At least one of the provided channel keys for poller initialization is invalid";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "shmem: at least one of the provided channel keys for poller initialization is invalid";
LOG(ERROR) << "shmem: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}
@ -153,7 +153,7 @@ FairMQPollerSHM::FairMQPollerSHM(const FairMQSocket& cmdSocket, const FairMQSock
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "shmem: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -164,11 +164,12 @@ void FairMQPollerSHM::Poll(const int timeout)
{
if (errno == ETERM)
{
LOG(DEBUG) << "polling exited, reason: " << zmq_strerror(errno);
LOG(DEBUG) << "shmem: polling exited, reason: " << zmq_strerror(errno);
}
else
{
LOG(ERROR) << "polling failed, reason: " << zmq_strerror(errno);
LOG(ERROR) << "shmem: polling failed, reason: " << zmq_strerror(errno);
throw std::runtime_error("shmem: polling failed");
}
}
}
@ -206,8 +207,8 @@ bool FairMQPollerSHM::CheckInput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "shmem: invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "shmem: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}
@ -225,8 +226,8 @@ bool FairMQPollerSHM::CheckOutput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "shmem: Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "shmem: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}

104
fairmq/shmem/FairMQShmManager.h
View File

@ -121,62 +121,68 @@ class Manager
bipc::managed_shared_memory* fSegment;
};
class Chunk
struct alignas(16) MetaHeader
{
public:
Chunk()
: fHandle()
, fSize(0)
{
}
Chunk(const size_t size)
: fHandle()
, fSize(size)
{
void* ptr = Manager::Instance().Segment()->allocate(size);
fHandle = Manager::Instance().Segment()->get_handle_from_address(ptr);
}
~Chunk()
{
Manager::Instance().Segment()->deallocate(Manager::Instance().Segment()->get_address_from_handle(fHandle));
}
// bipc::managed_shared_memory::handle_t GetHandle() const
// {
// return fHandle;
// }
void* GetData() const
{
return Manager::Instance().Segment()->get_address_from_handle(fHandle);
}
size_t GetSize() const
{
return fSize;
}
private:
uint64_t fSize;
bipc::managed_shared_memory::handle_t fHandle;
size_t fSize;
};
typedef bipc::managed_shared_ptr<Chunk, bipc::managed_shared_memory>::type ShPtrType;
// class Chunk
// {
// public:
// Chunk()
// : fHandle()
// , fSize(0)
// {
// }
struct ShPtrOwner
{
ShPtrOwner(const ShPtrType& other)
: fPtr(other)
{}
// Chunk(const size_t size)
// : fHandle()
// , fSize(size)
// {
// void* ptr = Manager::Instance().Segment()->allocate(size);
// fHandle = Manager::Instance().Segment()->get_handle_from_address(ptr);
// }
ShPtrOwner(const ShPtrOwner& other)
: fPtr(other.fPtr)
{}
// ~Chunk()
// {
// Manager::Instance().Segment()->deallocate(Manager::Instance().Segment()->get_address_from_handle(fHandle));
// }
ShPtrType fPtr;
};
// bipc::managed_shared_memory::handle_t GetHandle() const
// {
// return fHandle;
// }
// void* GetData() const
// {
// return Manager::Instance().Segment()->get_address_from_handle(fHandle);
// }
// size_t GetSize() const
// {
// return fSize;
// }
// private:
// bipc::managed_shared_memory::handle_t fHandle;
// size_t fSize;
// };
// typedef bipc::managed_shared_ptr<Chunk, bipc::managed_shared_memory>::type ShPtrType;
// struct ShPtrOwner
// {
// ShPtrOwner(const ShPtrType& other)
// : fPtr(other)
// {}
// ShPtrOwner(const ShPtrOwner& other)
// : fPtr(other.fPtr)
// {}
// ShPtrType fPtr;
// };
} // namespace shmem

406
fairmq/shmem/FairMQSocketSHM.cxx
View File

@ -19,6 +19,7 @@ using namespace FairMQ::shmem;
// Context to hold the ZeroMQ sockets
unique_ptr<FairMQContextSHM> FairMQSocketSHM::fContext; // = unique_ptr<FairMQContextSHM>(new FairMQContextSHM(1));
bool FairMQSocketSHM::fContextInitialized = false;
atomic<bool> FairMQSocketSHM::fInterrupted(false);
FairMQSocketSHM::FairMQSocketSHM(const string& type, const string& name, const int numIoThreads, const string& id /*= ""*/)
: FairMQSocket(ZMQ_SNDMORE, ZMQ_RCVMORE, ZMQ_DONTWAIT)
@ -57,22 +58,22 @@ FairMQSocketSHM::FairMQSocketSHM(const string& type, const string& name, const i
// Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating.
// Default value for ZeroMQ is -1, which is to wait forever.
int linger = 500;
int linger = 1000;
if (zmq_setsockopt(fSocket, ZMQ_LINGER, &linger, sizeof(linger)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
}
int kernelSndSize = 10000;
if (zmq_setsockopt(fSocket, ZMQ_SNDBUF, &kernelSndSize, sizeof(kernelSndSize)) != 0)
int sndTimeout = 700;
if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &sndTimeout, sizeof(sndTimeout)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_SNDBUF socket option, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);
}
int kernelRcvSize = 10000;
if (zmq_setsockopt(fSocket, ZMQ_RCVBUF, &kernelRcvSize, sizeof(kernelRcvSize)) != 0)
int rcvTimeout = 700;
if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &rcvTimeout, sizeof(rcvTimeout)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_RCVBUF socket option, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);
}
if (type == "sub")
@ -119,105 +120,183 @@ void FairMQSocketSHM::Connect(const string& address)
}
}
int FairMQSocketSHM::Send(FairMQMessage* msg, const string& flag)
int FairMQSocketSHM::Send(FairMQMessagePtr& msg, const int flags)
{
return Send(msg, GetConstant(flag));
}
int FairMQSocketSHM::Send(FairMQMessage* msg, const int flags)
{
int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, flags);
if (nbytes >= 0)
int nbytes = -1;
while (true && !fInterrupted)
{
static_cast<FairMQMessageSHM*>(msg)->fReceiving = false;
static_cast<FairMQMessageSHM*>(msg)->fQueued = true;
size_t size = msg->GetSize();
fBytesTx += size;
++fMessagesTx;
return size;
}
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
int64_t FairMQSocketSHM::Send(const vector<FairMQMessagePtr>& msgVec, const int flags)
{
// Sending vector typicaly handles more then one part
if (msgVec.size() > 1)
{
int64_t totalSize = 0;
for (unsigned int i = 0; i < msgVec.size() - 1; ++i)
nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, flags);
if (nbytes == 0)
{
int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msgVec[i]->GetMessage()), fSocket, ZMQ_SNDMORE|flags);
if (nbytes >= 0)
{
static_cast<FairMQMessageSHM*>(msgVec[i].get())->fReceiving = false;
static_cast<FairMQMessageSHM*>(msgVec[i].get())->fQueued = true;
size_t size = msgVec[i]->GetSize();
return nbytes;
}
else if (nbytes > 0)
{
// static_cast<FairMQMessageSHM*>(msg.get())->fReceiving = false;
static_cast<FairMQMessageSHM*>(msg.get())->fQueued = true;
totalSize += size;
fBytesTx += size;
size_t size = msg->GetSize();
fBytesTx += size;
++fMessagesTx;
return size;
}
else if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
continue;
}
else
{
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
return -2;
}
}
int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msgVec.back()->GetMessage()), fSocket, flags);
if (nbytes >= 0)
else if (zmq_errno() == ETERM)
{
static_cast<FairMQMessageSHM*>(msgVec.back().get())->fReceiving = false;
static_cast<FairMQMessageSHM*>(msgVec.back().get())->fQueued = true;
size_t size = msgVec.back()->GetSize();
totalSize += size;
fBytesTx += size;
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
// store statistics on how many messages have been sent (handle all parts as a single message)
++fMessagesTx;
return totalSize;
} // If there's only one part, send it as a regular message
else if (msgVec.size() == 1)
return -1;
}
int FairMQSocketSHM::Receive(FairMQMessagePtr& msg, const int flags)
{
int nbytes = -1;
zmq_msg_t* msgPtr = static_cast<zmq_msg_t*>(msg->GetMessage());
while (true)
{
return Send(msgVec.back().get(), flags);
nbytes = zmq_msg_recv(msgPtr, fSocket, flags);
if (nbytes == 0)
{
++fMessagesRx;
return nbytes;
}
else if (nbytes > 0)
{
// string ownerID(static_cast<char*>(zmq_msg_data(msgPtr)), zmq_msg_size(msgPtr));
// ShPtrOwner* owner = Manager::Instance().Segment()->find<ShPtrOwner>(ownerID.c_str()).first;
MetaHeader* hdr = static_cast<MetaHeader*>(zmq_msg_data(msgPtr));
size_t size = 0;
if (hdr->fHandle)
{
static_cast<FairMQMessageSHM*>(msg.get())->fHandle = hdr->fHandle;
static_cast<FairMQMessageSHM*>(msg.get())->fChunkSize = hdr->fSize;
// static_cast<FairMQMessageSHM*>(msg.get())->fOwner = owner;
// static_cast<FairMQMessageSHM*>(msg.get())->fReceiving = true;
size = msg->GetSize();
fBytesRx += size;
++fMessagesRx;
return size;
}
else
{
LOG(ERROR) << "Received meta data, but could not find corresponding chunk";
return -1;
}
}
else if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
continue;
}
else
{
return -2;
}
}
else if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
}
int64_t FairMQSocketSHM::Send(vector<FairMQMessagePtr>& msgVec, const int flags)
{
const unsigned int vecSize = msgVec.size();
// Sending vector typicaly handles more then one part
if (vecSize > 1)
{
int64_t totalSize = 0;
int nbytes = -1;
bool repeat = false;
while (true && !fInterrupted)
{
for (unsigned int i = 0; i < vecSize; ++i)
{
nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msgVec[i]->GetMessage()),
fSocket,
(i < vecSize - 1) ? ZMQ_SNDMORE|flags : flags);
if (nbytes >= 0)
{
static_cast<FairMQMessageSHM*>(msgVec[i].get())->fQueued = true;
// static_cast<FairMQMessageSHM*>(msgVec[i].get())->fReceiving = false;
// static_cast<FairMQMessageSHM*>(msgVec[i].get())->fQueued = true;
size_t size = msgVec[i]->GetSize();
totalSize += size;
}
else
{
// according to ZMQ docs, this can only occur for the first part
if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
repeat = true;
break;
}
else
{
return -2;
}
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
if (repeat)
{
continue;
}
// store statistics on how many messages have been sent (handle all parts as a single message)
++fMessagesTx;
fBytesTx += totalSize;
return totalSize;
}
return -1;
} // If there's only one part, send it as a regular message
else if (vecSize == 1)
{
return Send(msgVec.back(), flags);
}
else // if the vector is empty, something might be wrong
{
@ -226,112 +305,91 @@ int64_t FairMQSocketSHM::Send(const vector<FairMQMessagePtr>& msgVec, const int
}
}
int FairMQSocketSHM::Receive(FairMQMessage* msg, const string& flag)
{
return Receive(msg, GetConstant(flag));
}
int FairMQSocketSHM::Receive(FairMQMessage* msg, const int flags)
{
zmq_msg_t* msgPtr = static_cast<zmq_msg_t*>(msg->GetMessage());
int nbytes = zmq_msg_recv(msgPtr, fSocket, flags);
if (nbytes == 0)
{
++fMessagesRx;
return nbytes;
}
else if (nbytes > 0)
{
string ownerID(static_cast<char*>(zmq_msg_data(msgPtr)), zmq_msg_size(msgPtr));
ShPtrOwner* owner = Manager::Instance().Segment()->find<ShPtrOwner>(ownerID.c_str()).first;
size_t size = 0;
if (owner)
{
static_cast<FairMQMessageSHM*>(msg)->fOwner = owner;
static_cast<FairMQMessageSHM*>(msg)->fReceiving = true;
size = msg->GetSize();
fBytesRx += size;
++fMessagesRx;
return size;
}
else
{
LOG(ERROR) << "Received meta data, but could not find corresponding chunk";
return -1;
}
}
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int flags)
{
// Warn if the vector is filled before Receive() and empty it.
if (msgVec.size() > 0)
{
LOG(WARN) << "Message vector contains elements before Receive(), they will be deleted!";
msgVec.clear();
}
int64_t totalSize = 0;
int64_t more = 0;
bool repeat = false;
do
while (true)
{
FairMQMessagePtr part(new FairMQMessageSHM());
zmq_msg_t* msgPtr = static_cast<zmq_msg_t*>(part->GetMessage());
int nbytes = zmq_msg_recv(msgPtr, fSocket, flags);
if (nbytes == 0)
// Warn if the vector is filled before Receive() and empty it.
if (msgVec.size() > 0)
{
msgVec.push_back(move(part));
LOG(WARN) << "Message vector contains elements before Receive(), they will be deleted!";
msgVec.clear();
}
else if (nbytes > 0)
totalSize = 0;
more = 0;
repeat = false;
do
{
string ownerID(static_cast<char*>(zmq_msg_data(msgPtr)), zmq_msg_size(msgPtr));
ShPtrOwner* owner = Manager::Instance().Segment()->find<ShPtrOwner>(ownerID.c_str()).first;
size_t size = 0;
if (owner)
FairMQMessagePtr part(new FairMQMessageSHM());
zmq_msg_t* msgPtr = static_cast<zmq_msg_t*>(part->GetMessage());
int nbytes = zmq_msg_recv(msgPtr, fSocket, flags);
if (nbytes == 0)
{
static_cast<FairMQMessageSHM*>(part.get())->fOwner = owner;
static_cast<FairMQMessageSHM*>(part.get())->fReceiving = true;
size = part->GetSize();
msgVec.push_back(move(part));
}
else if (nbytes > 0)
{
// string ownerID(static_cast<char*>(zmq_msg_data(msgPtr)), zmq_msg_size(msgPtr));
// ShPtrOwner* owner = Manager::Instance().Segment()->find<ShPtrOwner>(ownerID.c_str()).first;
MetaHeader* hdr = static_cast<MetaHeader*>(zmq_msg_data(msgPtr));
size_t size = 0;
if (hdr->fHandle)
{
static_cast<FairMQMessageSHM*>(part.get())->fHandle = hdr->fHandle;
static_cast<FairMQMessageSHM*>(part.get())->fChunkSize = hdr->fSize;
// static_cast<FairMQMessageSHM*>(msg.get())->fOwner = owner;
// static_cast<FairMQMessageSHM*>(msg.get())->fReceiving = true;
size = part->GetSize();
fBytesRx += size;
totalSize += size;
msgVec.push_back(move(part));
totalSize += size;
}
else
{
LOG(ERROR) << "Received meta data, but could not find corresponding chunk";
return -1;
}
}
else if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
repeat = true;
break;
}
else
{
return -2;
}
}
else
{
LOG(ERROR) << "Received meta data, but could not find corresponding chunk";
return -1;
return nbytes;
}
size_t more_size = sizeof(more);
zmq_getsockopt(fSocket, ZMQ_RCVMORE, &more, &more_size);
}
else
while (more);
if (repeat)
{
return nbytes;
continue;
}
size_t more_size = sizeof(more);
zmq_getsockopt(fSocket, ZMQ_RCVMORE, &more, &more_size);
// store statistics on how many messages have been received (handle all parts as a single message)
++fMessagesRx;
fBytesRx += totalSize;
return totalSize;
}
while (more);
// store statistics on how many messages have been received (handle all parts as a single message)
++fMessagesRx;
return totalSize;
}
void FairMQSocketSHM::Close()
@ -362,11 +420,13 @@ void FairMQSocketSHM::Terminate()
void FairMQSocketSHM::Interrupt()
{
FairMQMessageSHM::fInterrupted = true;
fInterrupted = true;
}
void FairMQSocketSHM::Resume()
{
FairMQMessageSHM::fInterrupted = false;
fInterrupted = false;
}
void* FairMQSocketSHM::GetSocket() const

10
fairmq/shmem/FairMQSocketSHM.h
View File

@ -13,6 +13,7 @@
#include <memory> // unique_ptr
#include "FairMQSocket.h"
#include "FairMQMessage.h"
#include "FairMQContextSHM.h"
#include "FairMQShmManager.h"
@ -28,12 +29,10 @@ class FairMQSocketSHM : public FairMQSocket
virtual bool Bind(const std::string& address);
virtual void Connect(const std::string& address);
virtual int Send(FairMQMessage* msg, const std::string& flag = "");
virtual int Send(FairMQMessage* msg, const int flags = 0);
virtual int64_t Send(const std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual int Send(FairMQMessagePtr& msg, const int flags = 0);
virtual int Receive(FairMQMessagePtr& msg, const int flags = 0);
virtual int Receive(FairMQMessage* msg, const std::string& flag = "");
virtual int Receive(FairMQMessage* msg, const int flags = 0);
virtual int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual void* GetSocket() const;
@ -71,6 +70,7 @@ class FairMQSocketSHM : public FairMQSocket
static std::unique_ptr<FairMQContextSHM> fContext;
static bool fContextInitialized;
static std::atomic<bool> fInterrupted;
};
#endif /* FAIRMQSOCKETSHM_H_ */

10
fairmq/shmem/FairMQTransportFactorySHM.cxx
View File

@ -12,11 +12,13 @@
using namespace std;
static FairMQ::Transport gTransportType = FairMQ::Transport::SHM;
FairMQTransportFactorySHM::FairMQTransportFactorySHM()
{
int major, minor, patch;
zmq_version(&major, &minor, &patch);
LOG(DEBUG) << "Using ZeroMQ (" << major << "." << minor << "." << patch << ") & "
LOG(DEBUG) << "Transport: Using ZeroMQ (" << major << "." << minor << "." << patch << ") & "
<< "boost::interprocess (" << (BOOST_VERSION / 100000) << "." << (BOOST_VERSION / 100 % 1000) << "." << (BOOST_VERSION % 100) << ")";
}
@ -54,3 +56,9 @@ FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const FairMQSocket& cmdS
{
return unique_ptr<FairMQPoller>(new FairMQPollerSHM(cmdSocket, dataSocket));
}
FairMQ::Transport FairMQTransportFactorySHM::GetType() const
{
return gTransportType;
}

3
fairmq/shmem/FairMQTransportFactorySHM.h
View File

@ -9,6 +9,7 @@
#define FAIRMQTRANSPORTFACTORYSHM_H_
#include <vector>
#include <string>
#include "FairMQTransportFactory.h"
#include "FairMQContextSHM.h"
@ -31,6 +32,8 @@ class FairMQTransportFactorySHM : public FairMQTransportFactory
virtual FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const;
virtual FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const;
virtual FairMQ::Transport GetType() const;
virtual ~FairMQTransportFactorySHM() {};
};

7
fairmq/zeromq/FairMQMessageZMQ.cxx
View File

@ -20,6 +20,8 @@
using namespace std;
static FairMQ::Transport gTransportType = FairMQ::Transport::ZMQ;
string FairMQMessageZMQ::fDeviceID = string();
FairMQMessageZMQ::FairMQMessageZMQ()
@ -101,6 +103,11 @@ void FairMQMessageZMQ::SetDeviceId(const string& deviceId)
fDeviceID = deviceId;
}
FairMQ::Transport FairMQMessageZMQ::GetType() const
{
return gTransportType;
}
void FairMQMessageZMQ::Copy(const unique_ptr<FairMQMessage>& msg)
{
// Shares the message buffer between msg and this fMessage.

2
fairmq/zeromq/FairMQMessageZMQ.h
View File

@ -41,6 +41,8 @@ class FairMQMessageZMQ : public FairMQMessage
virtual void SetDeviceId(const std::string& deviceId);
virtual FairMQ::Transport GetType() const;
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg);
void CloseMessage();

25
fairmq/zeromq/FairMQPollerZMQ.cxx
View File

@ -51,7 +51,7 @@ FairMQPollerZMQ::FairMQPollerZMQ(const vector<FairMQChannel>& channels)
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "zeromq: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -105,7 +105,7 @@ FairMQPollerZMQ::FairMQPollerZMQ(const unordered_map<string, vector<FairMQChanne
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "zeromq: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -113,9 +113,9 @@ FairMQPollerZMQ::FairMQPollerZMQ(const unordered_map<string, vector<FairMQChanne
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "At least one of the provided channel keys for poller initialization is invalid";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
throw std::out_of_range("Invalid channel during poller initialization");
LOG(ERROR) << "zeromq: at least one of the provided channel keys for poller initialization is invalid";
LOG(ERROR) << "zeromq: out of range error: " << oor.what() << '\n';
throw std::out_of_range("invalid channel during poller initialization");
}
}
@ -153,7 +153,7 @@ FairMQPollerZMQ::FairMQPollerZMQ(const FairMQSocket& cmdSocket, const FairMQSock
}
else
{
LOG(ERROR) << "invalid poller configuration, exiting.";
LOG(ERROR) << "zeromq: invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@ -164,11 +164,12 @@ void FairMQPollerZMQ::Poll(const int timeout)
{
if (errno == ETERM)
{
LOG(DEBUG) << "polling exited, reason: " << zmq_strerror(errno);
LOG(DEBUG) << "zeromq: polling exited, reason: " << zmq_strerror(errno);
}
else
{
LOG(ERROR) << "polling failed, reason: " << zmq_strerror(errno);
LOG(ERROR) << "zeromq: polling failed, reason: " << zmq_strerror(errno);
throw std::runtime_error("zeromq: polling failed");
}
}
}
@ -206,8 +207,8 @@ bool FairMQPollerZMQ::CheckInput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "zeromq: invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "zeromq: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}
@ -225,8 +226,8 @@ bool FairMQPollerZMQ::CheckOutput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "Out of Range error: " << oor.what() << '\n';
LOG(ERROR) << "zeromq: invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "zeromq: out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}

313
fairmq/zeromq/FairMQSocketZMQ.cxx
View File

@ -23,7 +23,8 @@
using namespace std;
// Context to hold the ZeroMQ sockets
std::unique_ptr<FairMQContextZMQ> FairMQSocketZMQ::fContext = std::unique_ptr<FairMQContextZMQ>(new FairMQContextZMQ(1));
unique_ptr<FairMQContextZMQ> FairMQSocketZMQ::fContext = unique_ptr<FairMQContextZMQ>(new FairMQContextZMQ(1));
atomic<bool> FairMQSocketZMQ::fInterrupted(false);
FairMQSocketZMQ::FairMQSocketZMQ(const string& type, const string& name, const int numIoThreads, const string& id /*= ""*/)
: FairMQSocket(ZMQ_SNDMORE, ZMQ_RCVMORE, ZMQ_DONTWAIT)
@ -56,12 +57,24 @@ FairMQSocketZMQ::FairMQSocketZMQ(const string& type, const string& name, const i
// Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating.
// Default value for ZeroMQ is -1, which is to wait forever.
int linger = 500;
int linger = 1000;
if (zmq_setsockopt(fSocket, ZMQ_LINGER, &linger, sizeof(linger)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
}
int sndTimeout = 700;
if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &sndTimeout, sizeof(sndTimeout)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);
}
int rcvTimeout = 700;
if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &rcvTimeout, sizeof(rcvTimeout)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);
}
if (type == "sub")
{
if (zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, NULL, 0) != 0)
@ -106,91 +119,145 @@ void FairMQSocketZMQ::Connect(const string& address)
}
}
int FairMQSocketZMQ::Send(FairMQMessage* msg, const string& flag)
int FairMQSocketZMQ::Send(FairMQMessagePtr& msg, const int flags)
{
return Send(msg, GetConstant(flag));
}
int nbytes = -1;
int FairMQSocketZMQ::Send(FairMQMessage* msg, const int flags)
{
int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, flags);
if (nbytes >= 0)
while (true)
{
fBytesTx += nbytes;
++fMessagesTx;
return nbytes;
}
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
int64_t FairMQSocketZMQ::Send(const vector<unique_ptr<FairMQMessage>>& msgVec, const int flags)
{
// Sending vector typicaly handles more then one part
if (msgVec.size() > 1)
{
int64_t totalSize = 0;
for (unsigned int i = 0; i < msgVec.size() - 1; ++i)
nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, flags);
if (nbytes >= 0)
{
int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msgVec[i]->GetMessage()), fSocket, ZMQ_SNDMORE|flags);
if (nbytes >= 0)
fBytesTx += nbytes;
++fMessagesTx;
return nbytes;
}
else if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
totalSize += nbytes;
continue;
}
else
{
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
return -2;
}
}
int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msgVec.back()->GetMessage()), fSocket, flags);
if (nbytes >= 0)
else if (zmq_errno() == ETERM)
{
totalSize += nbytes;
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
}
// store statistics on how many messages have been sent (handle all parts as a single message)
++fMessagesTx;
fBytesTx += totalSize;
return totalSize;
} // If there's only one part, send it as a regular message
else if (msgVec.size() == 1)
int FairMQSocketZMQ::Receive(FairMQMessagePtr& msg, const int flags)
{
int nbytes = -1;
while (true)
{
return Send(msgVec.back().get(), flags);
nbytes = zmq_msg_recv(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, flags);
if (nbytes >= 0)
{
fBytesRx += nbytes;
++fMessagesRx;
return nbytes;
}
else if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
continue;
}
else
{
return -2;
}
}
else if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
}
int64_t FairMQSocketZMQ::Send(vector<unique_ptr<FairMQMessage>>& msgVec, const int flags)
{
const unsigned int vecSize = msgVec.size();
// Sending vector typicaly handles more then one part
if (vecSize > 1)
{
int64_t totalSize = 0;
int nbytes = -1;
bool repeat = false;
while (true)
{
totalSize = 0;
nbytes = -1;
repeat = false;
for (unsigned int i = 0; i < vecSize; ++i)
{
nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msgVec[i]->GetMessage()),
fSocket,
(i < vecSize - 1) ? ZMQ_SNDMORE|flags : flags);
if (nbytes >= 0)
{
totalSize += nbytes;
}
else
{
// according to ZMQ docs, this can only occur for the first part
if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
repeat = true;
break;
}
else
{
return -2;
}
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
if (repeat)
{
continue;
}
// store statistics on how many messages have been sent (handle all parts as a single message)
++fMessagesTx;
fBytesTx += totalSize;
return totalSize;
}
} // If there's only one part, send it as a regular message
else if (vecSize == 1)
{
return Send(msgVec.back(), flags);
}
else // if the vector is empty, something might be wrong
{
@ -199,69 +266,67 @@ int64_t FairMQSocketZMQ::Send(const vector<unique_ptr<FairMQMessage>>& msgVec, c
}
}
int FairMQSocketZMQ::Receive(FairMQMessage* msg, const string& flag)
{
return Receive(msg, GetConstant(flag));
}
int FairMQSocketZMQ::Receive(FairMQMessage* msg, const int flags)
{
int nbytes = zmq_msg_recv(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, flags);
if (nbytes >= 0)
{
fBytesRx += nbytes;
++fMessagesRx;
return nbytes;
}
if (zmq_errno() == EAGAIN)
{
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
int64_t FairMQSocketZMQ::Receive(vector<unique_ptr<FairMQMessage>>& msgVec, const int flags)
{
// Warn if the vector is filled before Receive() and empty it.
if (msgVec.size() > 0)
{
LOG(WARN) << "Message vector contains elements before Receive(), they will be deleted!";
msgVec.clear();
}
int64_t totalSize = 0;
int64_t more = 0;
bool repeat = false;
do
while (true)
{
unique_ptr<FairMQMessage> part(new FairMQMessageZMQ());
int nbytes = zmq_msg_recv(static_cast<zmq_msg_t*>(part->GetMessage()), fSocket, flags);
if (nbytes >= 0)
// Warn if the vector is filled before Receive() and empty it.
if (msgVec.size() > 0)
{
msgVec.push_back(move(part));
totalSize += nbytes;
}
else
{
return nbytes;
LOG(WARN) << "Message vector contains elements before Receive(), they will be deleted!";
msgVec.clear();
}
size_t more_size = sizeof(more);
zmq_getsockopt(fSocket, ZMQ_RCVMORE, &more, &more_size);
totalSize = 0;
more = 0;
repeat = false;
do
{
unique_ptr<FairMQMessage> part(new FairMQMessageZMQ());
int nbytes = zmq_msg_recv(static_cast<zmq_msg_t*>(part->GetMessage()), fSocket, flags);
if (nbytes >= 0)
{
msgVec.push_back(move(part));
totalSize += nbytes;
}
else if (zmq_errno() == EAGAIN)
{
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0))
{
repeat = true;
break;
}
else
{
return -2;
}
}
else
{
return nbytes;
}
size_t more_size = sizeof(more);
zmq_getsockopt(fSocket, ZMQ_RCVMORE, &more, &more_size);
}
while (more);
if (repeat)
{
continue;
}
// store statistics on how many messages have been received (handle all parts as a single message)
++fMessagesRx;
fBytesRx += totalSize;
return totalSize;
}
while (more);
// store statistics on how many messages have been received (handle all parts as a single message)
++fMessagesRx;
fBytesRx += totalSize;
return totalSize;
}
void FairMQSocketZMQ::Close()
@ -291,10 +356,12 @@ void FairMQSocketZMQ::Terminate()
void FairMQSocketZMQ::Interrupt()
{
fInterrupted = true;
}
void FairMQSocketZMQ::Resume()
{
fInterrupted = false;
}
void* FairMQSocketZMQ::GetSocket() const
@ -506,10 +573,6 @@ int FairMQSocketZMQ::GetConstant(const string& constant)
if (constant == "linger")
return ZMQ_LINGER;
if (constant == "no-block")
return ZMQ_DONTWAIT;
if (constant == "snd-more no-block")
return ZMQ_DONTWAIT|ZMQ_SNDMORE;
return -1;
}

10
fairmq/zeromq/FairMQSocketZMQ.h
View File

@ -20,6 +20,7 @@
#include <memory> // unique_ptr
#include "FairMQSocket.h"
#include "FairMQMessage.h"
#include "FairMQContextZMQ.h"
class FairMQSocketZMQ : public FairMQSocket
@ -34,12 +35,10 @@ class FairMQSocketZMQ : public FairMQSocket
virtual bool Bind(const std::string& address);
virtual void Connect(const std::string& address);
virtual int Send(FairMQMessage* msg, const std::string& flag = "");
virtual int Send(FairMQMessage* msg, const int flags = 0);
virtual int64_t Send(const std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual int Send(FairMQMessagePtr& msg, const int flags = 0);
virtual int Receive(FairMQMessagePtr& msg, const int flags = 0);
virtual int Receive(FairMQMessage* msg, const std::string& flag = "");
virtual int Receive(FairMQMessage* msg, const int flags = 0);
virtual int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0);
virtual void* GetSocket() const;
@ -76,6 +75,7 @@ class FairMQSocketZMQ : public FairMQSocket
std::atomic<unsigned long> fMessagesRx;
static std::unique_ptr<FairMQContextZMQ> fContext;
static std::atomic<bool> fInterrupted;
};
#endif /* FAIRMQSOCKETZMQ_H_ */

9
fairmq/zeromq/FairMQTransportFactoryZMQ.cxx
View File

@ -18,11 +18,13 @@
using namespace std;
static FairMQ::Transport gTransportType = FairMQ::Transport::ZMQ;
FairMQTransportFactoryZMQ::FairMQTransportFactoryZMQ()
{
int major, minor, patch;
zmq_version(&major, &minor, &patch);
LOG(DEBUG) << "Using ZeroMQ library, version: " << major << "." << minor << "." << patch;
LOG(DEBUG) << "Transport: Using ZeroMQ library, version: " << major << "." << minor << "." << patch;
}
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage() const
@ -59,3 +61,8 @@ FairMQPollerPtr FairMQTransportFactoryZMQ::CreatePoller(const FairMQSocket& cmdS
{
return unique_ptr<FairMQPoller>(new FairMQPollerZMQ(cmdSocket, dataSocket));
}
FairMQ::Transport FairMQTransportFactoryZMQ::GetType() const
{
return gTransportType;
}

3
fairmq/zeromq/FairMQTransportFactoryZMQ.h
View File

@ -16,6 +16,7 @@
#define FAIRMQTRANSPORTFACTORYZMQ_H_
#include <vector>
#include <string>
#include "FairMQTransportFactory.h"
#include "FairMQContextZMQ.h"
@ -38,6 +39,8 @@ class FairMQTransportFactoryZMQ : public FairMQTransportFactory
virtual FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const;
virtual FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const;
virtual FairMQ::Transport GetType() const;
virtual ~FairMQTransportFactoryZMQ() {};
};