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

Remove set/get timeout from general socket interface

Browse Source
This commit is contained in:
Alexey Rybalchenko 2018-10-12 17:36:54 +02:00 committed by Dennis Klein
parent e090967645
commit cfb727181f
9 changed files with 0 additions and 381 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
fairmq/FairMQSocket.h
View File

@ -48,11 +48,6 @@ class FairMQSocket
virtual unsigned long GetMessagesTx() const = 0; virtual unsigned long GetMessagesTx() const = 0;
virtual unsigned long GetMessagesRx() const = 0; virtual unsigned long GetMessagesRx() const = 0;
virtual bool SetSendTimeout(const int timeout, const std::string& address, const std::string& method) = 0;
virtual int GetSendTimeout() const = 0;
virtual bool SetReceiveTimeout(const int timeout, const std::string& address, const std::string& method) = 0;
virtual int GetReceiveTimeout() const = 0;
virtual ~FairMQSocket() {}; virtual ~FairMQSocket() {};
}; };

34
fairmq/nanomsg/FairMQSocketNN.cxx
View File

@ -492,40 +492,6 @@ unsigned long FairMQSocketNN::GetMessagesRx() const
return fMessagesRx; return fMessagesRx;
} }
bool FairMQSocketNN::SetSendTimeout(const int timeout, const string& /*address*/, const string& /*method*/)
{
fSndTimeout = timeout;
if (nn_setsockopt(fSocket, NN_SOL_SOCKET, NN_SNDTIMEO, &fSndTimeout, sizeof(fSndTimeout)) != 0)
{
LOG(error) << "Failed setting option 'send timeout' on socket " << fId << ", reason: " << nn_strerror(errno);
return false;
}
return true;
}
int FairMQSocketNN::GetSendTimeout() const
{
return fSndTimeout;
}
bool FairMQSocketNN::SetReceiveTimeout(const int timeout, const string& /*address*/, const string& /*method*/)
{
fRcvTimeout = timeout;
if (nn_setsockopt(fSocket, NN_SOL_SOCKET, NN_RCVTIMEO, &fRcvTimeout, sizeof(fRcvTimeout)) != 0)
{
LOG(error) << "Failed setting option 'receive timeout' on socket " << fId << ", reason: " << nn_strerror(errno);
return false;
}
return true;
}
int FairMQSocketNN::GetReceiveTimeout() const
{
return fRcvTimeout;
}
int FairMQSocketNN::GetConstant(const string& constant) int FairMQSocketNN::GetConstant(const string& constant)
{ {
if (constant == "") if (constant == "")

5
fairmq/nanomsg/FairMQSocketNN.h
View File

@ -53,11 +53,6 @@ class FairMQSocketNN : public FairMQSocket
unsigned long GetMessagesTx() const override; unsigned long GetMessagesTx() const override;
unsigned long GetMessagesRx() const override; unsigned long GetMessagesRx() const override;
bool SetSendTimeout(const int timeout, const std::string& address, const std::string& method) override;
int GetSendTimeout() const override;
bool SetReceiveTimeout(const int timeout, const std::string& address, const std::string& method) override;
int GetReceiveTimeout() const override;
static int GetConstant(const std::string& constant); static int GetConstant(const std::string& constant);
~FairMQSocketNN() override; ~FairMQSocketNN() override;

110
fairmq/ofi/Socket.cxx
View File

@ -619,116 +619,6 @@ auto Socket::GetOption(const string& option, void* value, size_t* valueSize) ->
} }
} }
auto Socket::SetSendTimeout(const int timeout, const string& address, const string& method) -> bool
{
throw SocketError{"Not yet implemented."};
// fSndTimeout = timeout;
// 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, &fSndTimeout, sizeof(fSndTimeout)) != 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, &fSndTimeout, sizeof(fSndTimeout)) != 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) << "timeout failed - unknown method provided!";
// return false;
// }
//
// return true;
}
auto Socket::GetSendTimeout() const -> int
{
throw SocketError{"Not yet implemented."};
// return fSndTimeout;
}
auto Socket::SetReceiveTimeout(const int timeout, const string& address, const string& method) -> bool
{
throw SocketError{"Not yet implemented."};
// fRcvTimeout = timeout;
// 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, &fRcvTimeout, sizeof(fRcvTimeout)) != 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, &fRcvTimeout, sizeof(fRcvTimeout)) != 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) << "timeout failed - unknown method provided!";
// return false;
// }
//
// return true;
}
auto Socket::GetReceiveTimeout() const -> int
{
throw SocketError{"Not yet implemented."};
// return fRcvTimeout;
}
auto Socket::GetConstant(const string& constant) -> int auto Socket::GetConstant(const string& constant) -> int
{ {
if (constant == "") if (constant == "")

5
fairmq/ofi/Socket.h
View File

@ -67,11 +67,6 @@ class Socket : public fair::mq::Socket
auto GetMessagesTx() const -> unsigned long override { return fMessagesTx; } auto GetMessagesTx() const -> unsigned long override { return fMessagesTx; }
auto GetMessagesRx() const -> unsigned long override { return fMessagesRx; } auto GetMessagesRx() const -> unsigned long override { return fMessagesRx; }
auto SetSendTimeout(const int timeout, const std::string& address, const std::string& method) -> bool override;
auto GetSendTimeout() const -> int override;
auto SetReceiveTimeout(const int timeout, const std::string& address, const std::string& method) -> bool override;
auto GetReceiveTimeout() const -> int override;
static auto GetConstant(const std::string& constant) -> int; static auto GetConstant(const std::string& constant) -> int;
~Socket() override; ~Socket() override;

106
fairmq/shmem/FairMQSocketSHM.cxx
View File

@ -497,112 +497,6 @@ unsigned long FairMQSocketSHM::GetMessagesRx() const
return fMessagesRx; return fMessagesRx;
} }
bool FairMQSocketSHM::SetSendTimeout(const int timeout, const string& address, const string& method)
{
fSndTimeout = timeout;
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, &fSndTimeout, sizeof(fSndTimeout)) != 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, &fSndTimeout, sizeof(fSndTimeout)) != 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) << "timeout failed - unknown method provided!";
return false;
}
return true;
}
int FairMQSocketSHM::GetSendTimeout() const
{
return fSndTimeout;
}
bool FairMQSocketSHM::SetReceiveTimeout(const int timeout, const string& address, const string& method)
{
fRcvTimeout = timeout;
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, &fRcvTimeout, sizeof(fRcvTimeout)) != 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, &fRcvTimeout, sizeof(fRcvTimeout)) != 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) << "timeout failed - unknown method provided!";
return false;
}
return true;
}
int FairMQSocketSHM::GetReceiveTimeout() const
{
return fRcvTimeout;
}
int FairMQSocketSHM::GetConstant(const string& constant) int FairMQSocketSHM::GetConstant(const string& constant)
{ {
if (constant == "") return 0; if (constant == "") return 0;

5
fairmq/shmem/FairMQSocketSHM.h
View File

@ -54,11 +54,6 @@ class FairMQSocketSHM : public FairMQSocket
unsigned long GetMessagesTx() const override; unsigned long GetMessagesTx() const override;
unsigned long GetMessagesRx() const override; unsigned long GetMessagesRx() const override;
bool SetSendTimeout(const int timeout, const std::string& address, const std::string& method) override;
int GetSendTimeout() const override;
bool SetReceiveTimeout(const int timeout, const std::string& address, const std::string& method) override;
int GetReceiveTimeout() const override;
static int GetConstant(const std::string& constant); static int GetConstant(const std::string& constant);
~FairMQSocketSHM() override; ~FairMQSocketSHM() override;

106
fairmq/zeromq/FairMQSocketZMQ.cxx
View File

@ -423,112 +423,6 @@ unsigned long FairMQSocketZMQ::GetMessagesRx() const
return fMessagesRx; return fMessagesRx;
} }
bool FairMQSocketZMQ::SetSendTimeout(const int timeout, const string& address, const string& method)
{
fSndTimeout = timeout;
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, &fSndTimeout, sizeof(fSndTimeout)) != 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, &fSndTimeout, sizeof(fSndTimeout)) != 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) << "timeout failed - unknown method provided!";
return false;
}
return true;
}
int FairMQSocketZMQ::GetSendTimeout() const
{
return fSndTimeout;
}
bool FairMQSocketZMQ::SetReceiveTimeout(const int timeout, const string& address, const string& method)
{
fRcvTimeout = timeout;
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, &fRcvTimeout, sizeof(fRcvTimeout)) != 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, &fRcvTimeout, sizeof(fRcvTimeout)) != 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) << "timeout failed - unknown method provided!";
return false;
}
return true;
}
int FairMQSocketZMQ::GetReceiveTimeout() const
{
return fRcvTimeout;
}
int FairMQSocketZMQ::GetConstant(const string& constant) int FairMQSocketZMQ::GetConstant(const string& constant)
{ {
if (constant == "") return 0; if (constant == "") return 0;

5
fairmq/zeromq/FairMQSocketZMQ.h
View File

@ -54,11 +54,6 @@ class FairMQSocketZMQ : public FairMQSocket
unsigned long GetMessagesTx() const override; unsigned long GetMessagesTx() const override;
unsigned long GetMessagesRx() const override; unsigned long GetMessagesRx() const override;
bool SetSendTimeout(const int timeout, const std::string& address, const std::string& method) override;
int GetSendTimeout() const override;
bool SetReceiveTimeout(const int timeout, const std::string& address, const std::string& method) override;
int GetReceiveTimeout() const override;
static int GetConstant(const std::string& constant); static int GetConstant(const std::string& constant);
~FairMQSocketZMQ() override; ~FairMQSocketZMQ() override;