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

Add methods to set timeout on blocking Send/Receive

Browse Source
This commit is contained in:
Alexey Rybalchenko
2015-10-15 16:43:30 +02:00
parent 91d3f013c0
commit 8f59db1283
11 changed files with 655 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
fairmq/zeromq/FairMQSocketZMQ.cxx
View File

@@ -37,20 +37,20 @@ FairMQSocketZMQ::FairMQSocketZMQ(const string& type, const string& name, int num
if (zmq_ctx_set(fContext->GetContext(), ZMQ_IO_THREADS, numIoThreads) != 0)
{
LOG(ERROR) << "failed configuring context, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed configuring context, reason: " << zmq_strerror(errno);
}
fSocket = zmq_socket(fContext->GetContext(), GetConstant(type));
if (fSocket == NULL)
{
LOG(ERROR) << "failed creating socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);
exit(EXIT_FAILURE);
}
if (zmq_setsockopt(fSocket, ZMQ_IDENTITY, &fId, fId.length()) != 0)
{
LOG(ERROR) << "failed setting ZMQ_IDENTITY socket option, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed setting ZMQ_IDENTITY socket option, reason: " << zmq_strerror(errno);
}
// Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating.
@@ -58,14 +58,14 @@ FairMQSocketZMQ::FairMQSocketZMQ(const string& type, const string& name, int num
int linger = 500;
if (zmq_setsockopt(fSocket, ZMQ_LINGER, &linger, sizeof(linger)) != 0)
{
LOG(ERROR) << "failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
}
if (type == "sub")
{
if (zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, NULL, 0) != 0)
{
LOG(ERROR) << "failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);
}
}
@@ -87,7 +87,7 @@ bool FairMQSocketZMQ::Bind(const string& address)
// do not print error in this case, this is handled by FairMQDevice in case no connection could be established after trying a number of random ports from a range.
return false;
}
LOG(ERROR) << "failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
return true;
@@ -99,7 +99,7 @@ void FairMQSocketZMQ::Connect(const string& address)
if (zmq_connect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
// error here means incorrect configuration. exit if it happens.
exit(EXIT_FAILURE);
}
@@ -116,14 +116,14 @@ int FairMQSocketZMQ::Send(FairMQMessage* msg, const string& flag)
}
if (zmq_errno() == EAGAIN)
{
return 0;
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
@@ -138,14 +138,14 @@ int FairMQSocketZMQ::Send(FairMQMessage* msg, const int flags)
}
if (zmq_errno() == EAGAIN)
{
return 0;
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
@@ -160,14 +160,14 @@ int FairMQSocketZMQ::Receive(FairMQMessage* msg, const string& flag)
}
if (zmq_errno() == EAGAIN)
{
return 0;
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
@@ -182,14 +182,14 @@ int FairMQSocketZMQ::Receive(FairMQMessage* msg, const int flags)
}
if (zmq_errno() == EAGAIN)
{
return 0;
return -2;
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
@@ -204,7 +204,7 @@ void FairMQSocketZMQ::Close()
if (zmq_close(fSocket) != 0)
{
LOG(ERROR) << "failed closing socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed closing socket " << fId << ", reason: " << zmq_strerror(errno);
}
fSocket = NULL;
@@ -214,16 +214,16 @@ void FairMQSocketZMQ::Terminate()
{
if (zmq_ctx_destroy(fContext->GetContext()) != 0)
{
LOG(ERROR) << "failed terminating context, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed terminating context, reason: " << zmq_strerror(errno);
}
}
void* FairMQSocketZMQ::GetSocket()
void* FairMQSocketZMQ::GetSocket() const
{
return fSocket;
}
int FairMQSocketZMQ::GetSocket(int nothing)
int FairMQSocketZMQ::GetSocket(int nothing) const
{
// dummy method to comply with the interface. functionality not possible in zeromq.
return -1;
@@ -233,7 +233,7 @@ void FairMQSocketZMQ::SetOption(const string& option, const void* value, size_t
{
if (zmq_setsockopt(fSocket, GetConstant(option), value, valueSize) < 0)
{
LOG(ERROR) << "failed setting socket option, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed setting socket option, reason: " << zmq_strerror(errno);
}
}
@@ -241,30 +241,150 @@ void FairMQSocketZMQ::GetOption(const string& option, void* value, size_t* value
{
if (zmq_getsockopt(fSocket, GetConstant(option), value, valueSize) < 0)
{
LOG(ERROR) << "failed getting socket option, reason: " << zmq_strerror(errno);
LOG(ERROR) << "Failed getting socket option, reason: " << zmq_strerror(errno);
}
}
unsigned long FairMQSocketZMQ::GetBytesTx()
unsigned long FairMQSocketZMQ::GetBytesTx() const
{
return fBytesTx;
}
unsigned long FairMQSocketZMQ::GetBytesRx()
unsigned long FairMQSocketZMQ::GetBytesRx() const
{
return fBytesRx;
}
unsigned long FairMQSocketZMQ::GetMessagesTx()
unsigned long FairMQSocketZMQ::GetMessagesTx() const
{
return fMessagesTx;
}
unsigned long FairMQSocketZMQ::GetMessagesRx()
unsigned long FairMQSocketZMQ::GetMessagesRx() const
{
return fMessagesRx;
}
bool FairMQSocketZMQ::SetSendTimeout(const int timeout, const string& address, const string& method)
{
if (method == "bind")
{
if (zmq_unbind(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed unbinding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_bind(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
else if (method == "connect")
{
if (zmq_disconnect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed disconnecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_connect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
else
{
LOG(ERROR) << "SetSendTimeout() failed - unknown method provided!";
return false;
}
return true;
}
int FairMQSocketZMQ::GetSendTimeout() const
{
int timeout = -1;
size_t size = sizeof(timeout);
if (zmq_getsockopt(fSocket, ZMQ_SNDTIMEO, &timeout, &size) != 0)
{
LOG(ERROR) << "Failed getting option 'receive timeout' on socket " << fId << ", reason: " << zmq_strerror(errno);
}
return timeout;
}
bool FairMQSocketZMQ::SetReceiveTimeout(const int timeout, const string& address, const string& method)
{
if (method == "bind")
{
if (zmq_unbind(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed unbinding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_bind(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
else if (method == "connect")
{
if (zmq_disconnect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed disconnecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_connect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
else
{
LOG(ERROR) << "SetReceiveTimeout() failed - unknown method provided!";
return false;
}
return true;
}
int FairMQSocketZMQ::GetReceiveTimeout() const
{
int timeout = -1;
size_t size = sizeof(timeout);
if (zmq_getsockopt(fSocket, ZMQ_RCVTIMEO, &timeout, &size) != 0)
{
LOG(ERROR) << "Failed getting option 'receive timeout' on socket " << fId << ", reason: " << zmq_strerror(errno);
}
return timeout;
}
int FairMQSocketZMQ::GetConstant(const string& constant)
{
if (constant == "")