Skip to content

Commit

Permalink
Add configurable default snd/rcv timeout
Browse files Browse the repository at this point in the history
  • Loading branch information
@rbx
rbx committed Dec 6, 2021
1 parent 856780f commit f24dee3
Show file tree
Hide file tree
Showing 6 changed files with 141 additions and 102 deletions.
10 changes: 10 additions & 0 deletions fairmq/Channel.cxx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ constexpr int Channel::DefaultSndBufSize;
constexpr int Channel::DefaultRcvBufSize;
constexpr int Channel::DefaultSndKernelSize;
constexpr int Channel::DefaultRcvKernelSize;
constexpr int Channel::DefaultSndTimeoutMs;
constexpr int Channel::DefaultRcvTimeoutMs;
constexpr int Channel::DefaultLinger;
constexpr int Channel::DefaultRateLogging;
constexpr int Channel::DefaultPortRangeMin;
Expand Down Expand Up @@ -73,6 +75,8 @@ Channel::Channel(string name, string type, string method, string address, shared
, fRcvBufSize(DefaultRcvBufSize)
, fSndKernelSize(DefaultSndKernelSize)
, fRcvKernelSize(DefaultRcvKernelSize)
, fSndTimeoutMs(DefaultSndTimeoutMs)
, fRcvTimeoutMs(DefaultRcvTimeoutMs)
, fLinger(DefaultLinger)
, fRateLogging(DefaultRateLogging)
, fPortRangeMin(DefaultPortRangeMin)
Expand All @@ -97,6 +101,8 @@ Channel::Channel(const string& name, int index, const Properties& properties)
fRcvBufSize = GetPropertyOrDefault(properties, string(prefix + "rcvBufSize"), DefaultRcvBufSize);
fSndKernelSize = GetPropertyOrDefault(properties, string(prefix + "sndKernelSize"), DefaultSndKernelSize);
fRcvKernelSize = GetPropertyOrDefault(properties, string(prefix + "rcvKernelSize"), DefaultRcvKernelSize);
fSndTimeoutMs = GetPropertyOrDefault(properties, string(prefix + "sndTimeoutMs"), DefaultSndTimeoutMs);
fRcvTimeoutMs = GetPropertyOrDefault(properties, string(prefix + "rcvTimeoutMs"), DefaultRcvTimeoutMs);
fLinger = GetPropertyOrDefault(properties, string(prefix + "linger"), DefaultLinger);
fRateLogging = GetPropertyOrDefault(properties, string(prefix + "rateLogging"), DefaultRateLogging);
fPortRangeMin = GetPropertyOrDefault(properties, string(prefix + "portRangeMin"), DefaultPortRangeMin);
Expand All @@ -120,6 +126,8 @@ Channel::Channel(const Channel& chan, string newName)
, fRcvBufSize(chan.fRcvBufSize)
, fSndKernelSize(chan.fSndKernelSize)
, fRcvKernelSize(chan.fRcvKernelSize)
, fSndTimeoutMs(chan.fSndTimeoutMs)
, fRcvTimeoutMs(chan.fRcvTimeoutMs)
, fLinger(chan.fLinger)
, fRateLogging(chan.fRateLogging)
, fPortRangeMin(chan.fPortRangeMin)
Expand All @@ -146,6 +154,8 @@ Channel& Channel::operator=(const Channel& chan)
fRcvBufSize = chan.fRcvBufSize;
fSndKernelSize = chan.fSndKernelSize;
fRcvKernelSize = chan.fRcvKernelSize;
fSndTimeoutMs = chan.fSndTimeoutMs;
fRcvTimeoutMs = chan.fRcvTimeoutMs;
fLinger = chan.fLinger;
fRateLogging = chan.fRateLogging;
fPortRangeMin = chan.fPortRangeMin;
Expand Down
112 changes: 62 additions & 50 deletions fairmq/Channel.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,14 @@ class Channel
/// @return Returns socket kernel transmit receive buffer size (in bytes)
int GetRcvKernelSize() const { return fRcvKernelSize; }

/// Get socket default send timeout (in ms)
/// @return Returns socket default send timeout (in ms)
int GetSndTimeout() const { return fSndTimeoutMs; }

/// Get socket default receive timeout (in ms)
/// @return Returns socket default receive timeout (in ms)
int GetRcvTimeout() const { return fRcvTimeoutMs; }

/// Get linger duration (in milliseconds)
/// @return Returns linger duration (in milliseconds)
int GetLinger() const { return fLinger; }
Expand Down Expand Up @@ -230,6 +238,14 @@ class Channel
/// @param rcvKernelSize Socket receive buffer size (in bytes)
void UpdateRcvKernelSize(int rcvKernelSize) { fRcvKernelSize = rcvKernelSize; Invalidate(); }

/// Set socket default send timeout (in ms)
/// @param sndTimeoutMs Socket default send timeout (in ms)
void UpdateSndTimeout(int sndTimeoutMs) { fSndTimeoutMs = sndTimeoutMs; Invalidate(); }

/// Set socket default receive timeout (in ms)
/// @param rcvTimeoutMs Socket default receive timeout (in ms)
void UpdateRcvTimeout(int rcvTimeoutMs) { fRcvTimeoutMs = rcvTimeoutMs; Invalidate(); }

/// Set linger duration (in milliseconds)
/// @param duration linger duration (in milliseconds)
void UpdateLinger(int duration) { fLinger = duration; Invalidate(); }
Expand Down Expand Up @@ -267,62 +283,52 @@ class Channel
/// invalidates the channel (requires validation to be used again).
void Invalidate() { fValid = false; }

/// Sends a message to the socket queue.
/// @param msg Constant reference of unique_ptr to a Message
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(MessagePtr& msg, int sndTimeoutInMs = -1)
{
CheckSendCompatibility(msg);
return fSocket->Send(msg, sndTimeoutInMs);
}

/// Receives a message from the socket queue.
/// @param msg Constant reference of unique_ptr to a Message
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(MessagePtr& msg, int rcvTimeoutInMs = -1)
{
CheckReceiveCompatibility(msg);
return fSocket->Receive(msg, rcvTimeoutInMs);
}

/// Send a vector of messages
/// @param msgVec message vector reference
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(std::vector<MessagePtr>& msgVec, int sndTimeoutInMs = -1)
/// Send message(s) to the socket queue.
/// @param m reference to MessagePtr/Parts/vector<MessagePtr>
/// @param sndTimeoutMs send timeout in ms.
/// -1 will wait forever (or until interrupt (e.g. via state change)),
/// 0 will not wait (return immediately if cannot send).
/// If not provided, default timeout will be taken.
/// @return Number of bytes that have been queued,
/// TransferCode::timeout if timed out,
/// TransferCode::error if there was an error,
/// TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M, typename... Timeout>
std::enable_if_t<is_transferrable<M>::value, int64_t>
Send(M& m, Timeout&&... sndTimeoutMs)
{
CheckSendCompatibility(msgVec);
return fSocket->Send(msgVec, sndTimeoutInMs);
}
static_assert(sizeof...(sndTimeoutMs) <= 1, "Send called with too many arguments");

/// Receive a vector of messages
/// @param msgVec message vector reference
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(std::vector<MessagePtr>& msgVec, int rcvTimeoutInMs = -1)
{
CheckReceiveCompatibility(msgVec);
return fSocket->Receive(msgVec, rcvTimeoutInMs);
CheckSendCompatibility(m);
int t = fSndTimeoutMs;
if constexpr (sizeof...(sndTimeoutMs) == 1) {
t = {sndTimeoutMs...};
}
return fSocket->Send(m, t);
}

/// Send Parts
/// @param parts Parts reference
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(Parts& parts, int sndTimeoutInMs = -1)
/// Receive message(s) from the socket queue.
/// @param m reference to MessagePtr/Parts/vector<MessagePtr>
/// @param rcvTimeoutMs receive timeout in ms.
/// -1 will wait forever (or until interrupt (e.g. via state change)),
/// 0 will not wait (return immediately if cannot receive).
/// If not provided, default timeout will be taken.
/// @return Number of bytes that have been received,
/// TransferCode::timeout if timed out,
/// TransferCode::error if there was an error,
/// TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M, typename... Timeout>
std::enable_if_t<is_transferrable<M>::value, int64_t>
Receive(M& m, Timeout&&... rcvTimeoutMs)
{
return Send(parts.fParts, sndTimeoutInMs);
}
static_assert(sizeof...(rcvTimeoutMs) <= 1, "Receive called with too many arguments");

/// Receive Parts
/// @param parts Parts reference
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(Parts& parts, int rcvTimeoutInMs = -1)
{
return Receive(parts.fParts, rcvTimeoutInMs);
CheckReceiveCompatibility(m);
int t = fRcvTimeoutMs;
if constexpr (sizeof...(rcvTimeoutMs) == 1) {
t = {rcvTimeoutMs...};
}
return fSocket->Receive(m, t);
}

unsigned long GetBytesTx() const { return fSocket->GetBytesTx(); }
Expand Down Expand Up @@ -366,6 +372,8 @@ class Channel
static constexpr int DefaultRcvBufSize = 1000;
static constexpr int DefaultSndKernelSize = 0;
static constexpr int DefaultRcvKernelSize = 0;
static constexpr int DefaultSndTimeoutMs = -1;
static constexpr int DefaultRcvTimeoutMs = -1;
static constexpr int DefaultLinger = 500;
static constexpr int DefaultRateLogging = 1;
static constexpr int DefaultPortRangeMin = 22000;
Expand All @@ -385,6 +393,8 @@ class Channel
int fRcvBufSize;
int fSndKernelSize;
int fRcvKernelSize;
int fSndTimeoutMs;
int fRcvTimeoutMs;
int fLinger;
int fRateLogging;
int fPortRangeMin;
Expand Down Expand Up @@ -414,6 +424,7 @@ class Channel
}
}

void CheckSendCompatibility(Parts& parts) { CheckSendCompatibility(parts.fParts); }
void CheckSendCompatibility(std::vector<MessagePtr>& msgVec)
{
for (auto& msg : msgVec) {
Expand Down Expand Up @@ -443,6 +454,7 @@ class Channel
}
}

void CheckReceiveCompatibility(Parts& parts) { CheckReceiveCompatibility(parts.fParts); }
void CheckReceiveCompatibility(std::vector<MessagePtr>& msgVec)
{
for (auto& msg : msgVec) {
Expand Down
102 changes: 50 additions & 52 deletions fairmq/Device.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,72 +81,70 @@ class Device
Deserializer().Deserialize(msg, std::forward<DataType>(data), std::forward<Args>(args)...);
}

/// Shorthand method to send `msg` on `chan` at index `i`
/// @param msg message reference
/// Send `m` on `chan` at index `i`
/// @param m reference to MessagePtr/Parts/vector<MessagePtr>
/// @param chan channel name
/// @param i channel index
/// @param sndTimeoutInMs send timeout in ms, -1 will wait forever (or until interrupt (e.g. via
/// state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out,
/// TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by
/// requested state change)
int64_t Send(MessagePtr& msg,
const std::string& channel,
const int index = 0,
int sndTimeoutInMs = -1)
/// @return Number of queued bytes,
/// TransferCode::timeout if timed out,
/// TransferCode::error if there was an error,
/// TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t>
Send(M& m, const std::string& channel, const int index = 0)
{
return GetChannel(channel, index).Send(msg, sndTimeoutInMs);
return GetChannel(channel, index).Send(m);
}

/// Shorthand method to receive `msg` on `chan` at index `i`
/// @param msg message reference
/// Receive `m` on `chan` at index `i`
/// @param m reference to MessagePtr/Parts/vector<MessagePtr>
/// @param chan channel name
/// @param i channel index
/// @param rcvTimeoutInMs receive timeout in ms, -1 will wait forever (or until interrupt (e.g.
/// via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out,
/// TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by
/// requested state change)
int64_t Receive(MessagePtr& msg,
const std::string& channel,
const int index = 0,
int rcvTimeoutInMs = -1)
/// @return Number of received bytes,
/// TransferCode::timeout if timed out,
/// TransferCode::error if there was an error,
/// TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t>
Receive(M& m, const std::string& channel, const int index = 0)
{
return GetChannel(channel, index).Receive(msg, rcvTimeoutInMs);
return GetChannel(channel, index).Receive(m);
}

/// Shorthand method to send Parts on `chan` at index `i`
/// @param parts parts reference
/// Send `m` on `chan` at index `i`
/// @param m reference to MessagePtr/Parts/vector<MessagePtr>
/// @param chan channel name
/// @param i channel index
/// @param sndTimeoutInMs send timeout in ms, -1 will wait forever (or until interrupt (e.g. via
/// state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out,
/// TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by
/// requested state change)
int64_t Send(Parts& parts,
const std::string& channel,
const int index = 0,
int sndTimeoutInMs = -1)
{
return GetChannel(channel, index).Send(parts.fParts, sndTimeoutInMs);
}

/// Shorthand method to receive Parts on `chan` at index `i`
/// @param parts parts reference
/// @param sndTimeoutMs send timeout in ms,
/// -1 will wait forever (or until interrupt (e.g. via state change)),
/// 0 will not wait (return immediately if cannot send)
/// @return Number of queued bytes,
/// TransferCode::timeout if timed out,
/// TransferCode::error if there was an error,
/// TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t>
Send(M& m, const std::string& channel, const int index, int sndTimeoutMs)
{
return GetChannel(channel, index).Send(m, sndTimeoutMs);
}

/// Receive `m` on `chan` at index `i`
/// @param m reference to MessagePtr/Parts/vector<MessagePtr>
/// @param chan channel name
/// @param i channel index
/// @param rcvTimeoutInMs receive timeout in ms, -1 will wait forever (or until interrupt (e.g.
/// via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out,
/// TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by
/// requested state change)
int64_t Receive(Parts& parts,
const std::string& channel,
const int index = 0,
int rcvTimeoutInMs = -1)
{
return GetChannel(channel, index).Receive(parts.fParts, rcvTimeoutInMs);
/// @param rcvTimeoutMs receive timeout in ms,
/// -1 will wait forever (or until interrupt (e.g. via state change),
/// 0 will not wait (return immediately if cannot receive)
/// @return Number of received bytes,
/// TransferCode::timeout if timed out,
/// TransferCode::error if there was an error,
/// TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t>
Receive(M& m, const std::string& channel, const int index, int rcvTimeoutMs)
{
return GetChannel(channel, index).Receive(m, rcvTimeoutMs);
}

/// @brief Getter for default transport factory
Expand Down
4 changes: 4 additions & 0 deletions fairmq/JSONParser.cxx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,8 @@ void ChannelParser(const ptree& tree, fair::mq::Properties& properties)
commonProperties.emplace("rcvBufSize", cn.second.get<int>("rcvBufSize", FairMQChannel::DefaultRcvBufSize));
commonProperties.emplace("sndKernelSize", cn.second.get<int>("sndKernelSize", FairMQChannel::DefaultSndKernelSize));
commonProperties.emplace("rcvKernelSize", cn.second.get<int>("rcvKernelSize", FairMQChannel::DefaultRcvKernelSize));
commonProperties.emplace("sndTimeoutMs", cn.second.get<int>("sndTimeoutMs", FairMQChannel::DefaultSndTimeoutMs));
commonProperties.emplace("rcvTimeoutMs", cn.second.get<int>("rcvTimeoutMs", FairMQChannel::DefaultRcvTimeoutMs));
commonProperties.emplace("linger", cn.second.get<int>("linger", FairMQChannel::DefaultLinger));
commonProperties.emplace("rateLogging", cn.second.get<int>("rateLogging", FairMQChannel::DefaultRateLogging));
commonProperties.emplace("portRangeMin", cn.second.get<int>("portRangeMin", FairMQChannel::DefaultPortRangeMin));
Expand Down Expand Up @@ -146,6 +148,8 @@ void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties
newProperties["rcvBufSize"] = sn.second.get<int>("rcvBufSize", boost::any_cast<int>(commonProperties.at("rcvBufSize")));
newProperties["sndKernelSize"] = sn.second.get<int>("sndKernelSize", boost::any_cast<int>(commonProperties.at("sndKernelSize")));
newProperties["rcvKernelSize"] = sn.second.get<int>("rcvKernelSize", boost::any_cast<int>(commonProperties.at("rcvKernelSize")));
newProperties["sndTimeoutMs"] = sn.second.get<int>("sndTimeoutMs", boost::any_cast<int>(commonProperties.at("sndTimeoutMs")));
newProperties["rcvTimeoutMs"] = sn.second.get<int>("rcvTimeoutMs", boost::any_cast<int>(commonProperties.at("rcvTimeoutMs")));
newProperties["linger"] = sn.second.get<int>("linger", boost::any_cast<int>(commonProperties.at("linger")));
newProperties["rateLogging"] = sn.second.get<int>("rateLogging", boost::any_cast<int>(commonProperties.at("rateLogging")));
newProperties["portRangeMin"] = sn.second.get<int>("portRangeMin", boost::any_cast<int>(commonProperties.at("portRangeMin")));
Expand Down
Loading

0 comments on commit f24dee3

Please sign in to comment.