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 thread-safety from channel methods

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-09-22 09:47:16 +02:00 committed by Dennis Klein
parent 3ab10ced7a
commit e332e20dbd
2 changed files with 58 additions and 196 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

239
fairmq/FairMQChannel.cxx
View File

@ -85,7 +85,6 @@ FairMQChannel::FairMQChannel(const string& name, const string& type, const strin
, fMultipart(false) , fMultipart(false)
, fModified(true) , fModified(true)
, fReset(false) , fReset(false)
, fMtx()
{} {}
FairMQChannel::FairMQChannel(const string& name, int index, const fair::mq::Properties& properties) FairMQChannel::FairMQChannel(const string& name, int index, const fair::mq::Properties& properties)
@ -141,33 +140,26 @@ FairMQChannel& FairMQChannel::operator=(const FairMQChannel& chan)
return *this; return *this;
} }
{ fTransportFactory = nullptr;
// TODO: replace this with std::scoped_lock (c++17) fTransportType = chan.fTransportType;
lock(fMtx, chan.fMtx); fSocket = nullptr;
lock_guard<mutex> lock1(fMtx, adopt_lock); fName = chan.fName;
lock_guard<mutex> lock2(chan.fMtx, adopt_lock); fType = chan.fType;
fMethod = chan.fMethod;
fTransportFactory = nullptr; fAddress = chan.fAddress;
fTransportType = chan.fTransportType; fSndBufSize = chan.fSndBufSize;
fSocket = nullptr; fRcvBufSize = chan.fRcvBufSize;
fName = chan.fName; fSndKernelSize = chan.fSndKernelSize;
fType = chan.fType; fRcvKernelSize = chan.fRcvKernelSize;
fMethod = chan.fMethod; fLinger = chan.fLinger;
fAddress = chan.fAddress; fRateLogging = chan.fRateLogging;
fSndBufSize = chan.fSndBufSize; fPortRangeMin = chan.fPortRangeMin;
fRcvBufSize = chan.fRcvBufSize; fPortRangeMax = chan.fPortRangeMax;
fSndKernelSize = chan.fSndKernelSize; fAutoBind = chan.fAutoBind;
fRcvKernelSize = chan.fRcvKernelSize; fIsValid = false;
fLinger = chan.fLinger; fMultipart = chan.fMultipart;
fRateLogging = chan.fRateLogging; fModified = chan.fModified;
fPortRangeMin = chan.fPortRangeMin; fReset = false;
fPortRangeMax = chan.fPortRangeMax;
fAutoBind = chan.fAutoBind;
fIsValid = false;
fMultipart = chan.fMultipart;
fModified = chan.fModified;
fReset = false;
}
return *this; return *this;
} }
@ -180,13 +172,11 @@ FairMQSocket & FairMQChannel::GetSocket() const
string FairMQChannel::GetName() const string FairMQChannel::GetName() const
{ {
lock_guard<mutex> lock(fMtx);
return fName; return fName;
} }
string FairMQChannel::GetPrefix() const string FairMQChannel::GetPrefix() const
{ {
lock_guard<mutex> lock(fMtx);
string prefix = fName; string prefix = fName;
prefix = prefix.erase(fName.rfind('[')); prefix = prefix.erase(fName.rfind('['));
return prefix; return prefix;
@ -194,7 +184,6 @@ string FairMQChannel::GetPrefix() const
string FairMQChannel::GetIndex() const string FairMQChannel::GetIndex() const
{ {
lock_guard<mutex> lock(fMtx);
string indexStr = fName; string indexStr = fName;
indexStr.erase(indexStr.rfind(']')); indexStr.erase(indexStr.rfind(']'));
indexStr.erase(0, indexStr.rfind('[') + 1); indexStr.erase(0, indexStr.rfind('[') + 1);
@ -202,307 +191,185 @@ string FairMQChannel::GetIndex() const
} }
string FairMQChannel::GetType() const string FairMQChannel::GetType() const
try { {
lock_guard<mutex> lock(fMtx);
return fType; return fType;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetType: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
string FairMQChannel::GetMethod() const string FairMQChannel::GetMethod() const
try { {
lock_guard<mutex> lock(fMtx);
return fMethod; return fMethod;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetMethod: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
string FairMQChannel::GetAddress() const string FairMQChannel::GetAddress() const
try { {
lock_guard<mutex> lock(fMtx);
return fAddress; return fAddress;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetAddress: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
string FairMQChannel::GetTransportName() const string FairMQChannel::GetTransportName() const
try { {
lock_guard<mutex> lock(fMtx);
return TransportNames.at(fTransportType); return TransportNames.at(fTransportType);
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetTransportName: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
Transport FairMQChannel::GetTransportType() const Transport FairMQChannel::GetTransportType() const
try { {
lock_guard<mutex> lock(fMtx);
return fTransportType; return fTransportType;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetTransportType: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetSndBufSize() const int FairMQChannel::GetSndBufSize() const
try { {
lock_guard<mutex> lock(fMtx);
return fSndBufSize; return fSndBufSize;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetSndBufSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetRcvBufSize() const int FairMQChannel::GetRcvBufSize() const
try { {
lock_guard<mutex> lock(fMtx);
return fRcvBufSize; return fRcvBufSize;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetRcvBufSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetSndKernelSize() const int FairMQChannel::GetSndKernelSize() const
try { {
lock_guard<mutex> lock(fMtx);
return fSndKernelSize; return fSndKernelSize;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetSndKernelSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetRcvKernelSize() const int FairMQChannel::GetRcvKernelSize() const
try { {
lock_guard<mutex> lock(fMtx);
return fRcvKernelSize; return fRcvKernelSize;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetRcvKernelSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetLinger() const int FairMQChannel::GetLinger() const
try { {
lock_guard<mutex> lock(fMtx);
return fLinger; return fLinger;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetLinger: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetRateLogging() const int FairMQChannel::GetRateLogging() const
try { {
lock_guard<mutex> lock(fMtx);
return fRateLogging; return fRateLogging;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetRateLogging: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetPortRangeMin() const int FairMQChannel::GetPortRangeMin() const
try { {
lock_guard<mutex> lock(fMtx);
return fPortRangeMin; return fPortRangeMin;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetPortRangeMin: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
int FairMQChannel::GetPortRangeMax() const int FairMQChannel::GetPortRangeMax() const
try { {
lock_guard<mutex> lock(fMtx);
return fPortRangeMax; return fPortRangeMax;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetPortRangeMax: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
bool FairMQChannel::GetAutoBind() const bool FairMQChannel::GetAutoBind() const
try { {
lock_guard<mutex> lock(fMtx);
return fAutoBind; return fAutoBind;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::GetAutoBind: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateType(const string& type) void FairMQChannel::UpdateType(const string& type)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fType = type; fType = type;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateType: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateMethod(const string& method) void FairMQChannel::UpdateMethod(const string& method)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fMethod = method; fMethod = method;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateMethod: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateAddress(const string& address) void FairMQChannel::UpdateAddress(const string& address)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fAddress = address; fAddress = address;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateAddress: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateTransport(const string& transport) void FairMQChannel::UpdateTransport(const string& transport)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fTransportType = TransportTypes.at(transport); fTransportType = TransportTypes.at(transport);
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateTransport: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateSndBufSize(const int sndBufSize) void FairMQChannel::UpdateSndBufSize(const int sndBufSize)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fSndBufSize = sndBufSize; fSndBufSize = sndBufSize;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateSndBufSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateRcvBufSize(const int rcvBufSize) void FairMQChannel::UpdateRcvBufSize(const int rcvBufSize)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fRcvBufSize = rcvBufSize; fRcvBufSize = rcvBufSize;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateRcvBufSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateSndKernelSize(const int sndKernelSize) void FairMQChannel::UpdateSndKernelSize(const int sndKernelSize)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fSndKernelSize = sndKernelSize; fSndKernelSize = sndKernelSize;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateSndKernelSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateRcvKernelSize(const int rcvKernelSize) void FairMQChannel::UpdateRcvKernelSize(const int rcvKernelSize)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fRcvKernelSize = rcvKernelSize; fRcvKernelSize = rcvKernelSize;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateRcvKernelSize: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateLinger(const int duration) void FairMQChannel::UpdateLinger(const int duration)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fLinger = duration; fLinger = duration;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateLinger: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateRateLogging(const int rateLogging) void FairMQChannel::UpdateRateLogging(const int rateLogging)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fRateLogging = rateLogging; fRateLogging = rateLogging;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateRateLogging: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdatePortRangeMin(const int minPort) void FairMQChannel::UpdatePortRangeMin(const int minPort)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fPortRangeMin = minPort; fPortRangeMin = minPort;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdatePortRangeMin: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdatePortRangeMax(const int maxPort) void FairMQChannel::UpdatePortRangeMax(const int maxPort)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fPortRangeMax = maxPort; fPortRangeMax = maxPort;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdatePortRangeMax: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateAutoBind(const bool autobind) void FairMQChannel::UpdateAutoBind(const bool autobind)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fAutoBind = autobind; fAutoBind = autobind;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateAutoBind: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
auto FairMQChannel::SetModified(const bool modified) -> void auto FairMQChannel::SetModified(const bool modified) -> void
try { {
lock_guard<mutex> lock(fMtx);
fModified = modified; fModified = modified;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::SetModified: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
void FairMQChannel::UpdateName(const string& name) void FairMQChannel::UpdateName(const string& name)
try { {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
fName = name; fName = name;
fModified = true; fModified = true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::UpdateName: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
bool FairMQChannel::IsValid() const bool FairMQChannel::IsValid() const
try { {
lock_guard<mutex> lock(fMtx);
return fIsValid; return fIsValid;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::IsValid: " << e.what();
throw ChannelConfigurationError(tools::ToString("failed to acquire lock: ", e.what()));
} }
bool FairMQChannel::Validate() bool FairMQChannel::Validate()
try { try {
lock_guard<mutex> lock(fMtx);
stringstream ss; stringstream ss;
ss << "Validating channel '" << fName << "'... "; ss << "Validating channel '" << fName << "'... ";
@ -652,8 +519,6 @@ try {
void FairMQChannel::Init() void FairMQChannel::Init()
{ {
lock_guard<mutex> lock(fMtx);
fSocket = fTransportFactory->CreateSocket(fType, fName); fSocket = fTransportFactory->CreateSocket(fType, fName);
// set linger duration (how long socket should wait for outstanding transfers before shutdown) // set linger duration (how long socket should wait for outstanding transfers before shutdown)
@ -674,14 +539,11 @@ void FairMQChannel::Init()
bool FairMQChannel::ConnectEndpoint(const string& endpoint) bool FairMQChannel::ConnectEndpoint(const string& endpoint)
{ {
lock_guard<mutex> lock(fMtx);
return fSocket->Connect(endpoint); return fSocket->Connect(endpoint);
} }
bool FairMQChannel::BindEndpoint(string& endpoint) bool FairMQChannel::BindEndpoint(string& endpoint)
{ {
lock_guard<mutex> lock(fMtx);
// try to bind to the configured port. If it fails, try random one (if AutoBind is on). // try to bind to the configured port. If it fails, try random one (if AutoBind is on).
if (fSocket->Bind(endpoint)) { if (fSocket->Bind(endpoint)) {
return true; return true;
@ -725,7 +587,6 @@ bool FairMQChannel::BindEndpoint(string& endpoint)
void FairMQChannel::ResetChannel() void FairMQChannel::ResetChannel()
{ {
lock_guard<mutex> lock(fMtx);
fIsValid = false; fIsValid = false;
// TODO: implement channel resetting // TODO: implement channel resetting
} }

15
fairmq/FairMQChannel.h
View File

@ -25,6 +25,12 @@
#include <utility> // std::move #include <utility> // std::move
#include <cstdint> // int64_t #include <cstdint> // int64_t
/**
* @class FairMQChannel FairMQChannel.h <FairMQChannel.h>
* @brief Wrapper class for FairMQSocket and related methods
*
* The class is not thread-safe.
*/
class FairMQChannel class FairMQChannel
{ {
friend class FairMQDevice; friend class FairMQDevice;
@ -100,7 +106,7 @@ class FairMQChannel
/// Get channel name /// Get channel name
/// @return Returns full channel name (e.g. "data[0]") /// @return Returns full channel name (e.g. "data[0]")
std::string GetName() const ; std::string GetName() const;
/// Get channel prefix /// Get channel prefix
/// @return Returns channel prefix (e.g. "data" in "data[0]") /// @return Returns channel prefix (e.g. "data" in "data[0]")
@ -302,10 +308,7 @@ class FairMQChannel
unsigned long GetMessagesTx() const { return fSocket->GetMessagesTx(); } unsigned long GetMessagesTx() const { return fSocket->GetMessagesTx(); }
unsigned long GetMessagesRx() const { return fSocket->GetMessagesRx(); } unsigned long GetMessagesRx() const { return fSocket->GetMessagesRx(); }
auto Transport() -> FairMQTransportFactory* auto Transport() -> FairMQTransportFactory* { return fTransportFactory.get(); };
{
return fTransportFactory.get();
};
template<typename... Args> template<typename... Args>
FairMQMessagePtr NewMessage(Args&&... args) FairMQMessagePtr NewMessage(Args&&... args)
@ -372,8 +375,6 @@ class FairMQChannel
bool fModified; bool fModified;
bool fReset; bool fReset;
mutable std::mutex fMtx;
void CheckSendCompatibility(FairMQMessagePtr& msg) void CheckSendCompatibility(FairMQMessagePtr& msg)
{ {
if (fTransportType != msg->GetType()) { if (fTransportType != msg->GetType()) {