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

Use exceptions for fatal errors in device/channel

Browse Source 
- These will be caught by StateMachine::ProcessWork
   and lead to error state.
 - Solve issue where device goes into ready state if
   it encounters misconfigured channel in the Run.
 - deprecate WaitForInitialValidation().
This commit is contained in:
Alexey Rybalchenko 2018-10-31 14:32:31 +01:00 committed by Dennis Klein
parent 3561255cf9
commit 3b5b2b501f
4 changed files with 397 additions and 530 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

768
fairmq/FairMQChannel.cxx
View File

@ -13,6 +13,7 @@
*/ */
#include "FairMQChannel.h" #include "FairMQChannel.h"
#include <fairmq/Tools.h>
#include <boost/algorithm/string.hpp> // join/split #include <boost/algorithm/string.hpp> // join/split
@ -156,524 +157,403 @@ string FairMQChannel::GetChannelIndex() const
} }
string FairMQChannel::GetType() const string FairMQChannel::GetType() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fType;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fType; LOG(error) << "Exception caught in FairMQChannel::GetType: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetType: " << e.what();
exit(EXIT_FAILURE);
}
} }
string FairMQChannel::GetMethod() const string FairMQChannel::GetMethod() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fMethod;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fMethod; LOG(error) << "Exception caught in FairMQChannel::GetMethod: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetMethod: " << e.what();
exit(EXIT_FAILURE);
}
} }
string FairMQChannel::GetAddress() const string FairMQChannel::GetAddress() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fAddress;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fAddress; LOG(error) << "Exception caught in FairMQChannel::GetAddress: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetAddress: " << e.what();
exit(EXIT_FAILURE);
}
} }
string FairMQChannel::GetTransportName() const string FairMQChannel::GetTransportName() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fair::mq::TransportNames.at(fTransportType);
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fair::mq::TransportNames.at(fTransportType); LOG(error) << "Exception caught in FairMQChannel::GetTransportName: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetTransportName: " << e.what();
exit(EXIT_FAILURE);
}
} }
int FairMQChannel::GetSndBufSize() const int FairMQChannel::GetSndBufSize() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fSndBufSize;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fSndBufSize; LOG(error) << "Exception caught in FairMQChannel::GetSndBufSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetSndBufSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
int FairMQChannel::GetRcvBufSize() const int FairMQChannel::GetRcvBufSize() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fRcvBufSize;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fRcvBufSize; LOG(error) << "Exception caught in FairMQChannel::GetRcvBufSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetRcvBufSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
int FairMQChannel::GetSndKernelSize() const int FairMQChannel::GetSndKernelSize() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fSndKernelSize;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fSndKernelSize; LOG(error) << "Exception caught in FairMQChannel::GetSndKernelSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetSndKernelSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
int FairMQChannel::GetRcvKernelSize() const int FairMQChannel::GetRcvKernelSize() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fRcvKernelSize;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fRcvKernelSize; LOG(error) << "Exception caught in FairMQChannel::GetRcvKernelSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetRcvKernelSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
int FairMQChannel::GetLinger() const int FairMQChannel::GetLinger() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fLinger;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fLinger; LOG(error) << "Exception caught in FairMQChannel::GetLinger: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetLinger: " << e.what();
exit(EXIT_FAILURE);
}
} }
int FairMQChannel::GetRateLogging() const int FairMQChannel::GetRateLogging() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fRateLogging;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fRateLogging; LOG(error) << "Exception caught in FairMQChannel::GetRateLogging: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::GetRateLogging: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateType(const string& type) void FairMQChannel::UpdateType(const string& type)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fType = type;
fIsValid = false; fModified = true;
fType = type; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateType: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateType: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateMethod(const string& method) void FairMQChannel::UpdateMethod(const string& method)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fMethod = method;
fIsValid = false; fModified = true;
fMethod = method; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateMethod: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateMethod: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateAddress(const string& address) void FairMQChannel::UpdateAddress(const string& address)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fAddress = address;
fIsValid = false; fModified = true;
fAddress = address; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateAddress: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateAddress: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateTransport(const string& transport) void FairMQChannel::UpdateTransport(const string& transport)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fTransportType = fair::mq::TransportTypes.at(transport);
fIsValid = false; fModified = true;
fTransportType = fair::mq::TransportTypes.at(transport); } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateTransport: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateTransport: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateSndBufSize(const int sndBufSize) void FairMQChannel::UpdateSndBufSize(const int sndBufSize)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fSndBufSize = sndBufSize;
fIsValid = false; fModified = true;
fSndBufSize = sndBufSize; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateSndBufSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateSndBufSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateRcvBufSize(const int rcvBufSize) void FairMQChannel::UpdateRcvBufSize(const int rcvBufSize)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fRcvBufSize = rcvBufSize;
fIsValid = false; fModified = true;
fRcvBufSize = rcvBufSize; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateRcvBufSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateRcvBufSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateSndKernelSize(const int sndKernelSize) void FairMQChannel::UpdateSndKernelSize(const int sndKernelSize)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fSndKernelSize = sndKernelSize;
fIsValid = false; fModified = true;
fSndKernelSize = sndKernelSize; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateSndKernelSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateSndKernelSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateRcvKernelSize(const int rcvKernelSize) void FairMQChannel::UpdateRcvKernelSize(const int rcvKernelSize)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fRcvKernelSize = rcvKernelSize;
fIsValid = false; fModified = true;
fRcvKernelSize = rcvKernelSize; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateRcvKernelSize: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateRcvKernelSize: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateLinger(const int duration) void FairMQChannel::UpdateLinger(const int duration)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fLinger = duration;
fIsValid = false; fModified = true;
fLinger = duration; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateLinger: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateLinger: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateRateLogging(const int rateLogging) void FairMQChannel::UpdateRateLogging(const int rateLogging)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fRateLogging = rateLogging;
fIsValid = false; fModified = true;
fRateLogging = rateLogging; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateRateLogging: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateRateLogging: " << e.what();
exit(EXIT_FAILURE);
}
} }
auto FairMQChannel::SetModified(const bool modified) -> void auto FairMQChannel::SetModified(const bool modified) -> void
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fModified = modified;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
fModified = modified; LOG(error) << "Exception caught in FairMQChannel::SetModified: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::SetModified: " << e.what();
exit(EXIT_FAILURE);
}
} }
void FairMQChannel::UpdateChannelName(const string& name) void FairMQChannel::UpdateChannelName(const string& name)
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ fIsValid = false;
unique_lock<mutex> lock(fChannelMutex); fName = name;
fIsValid = false; fModified = true;
fName = name; } catch (exception& e) {
fModified = true; LOG(error) << "Exception caught in FairMQChannel::UpdateChannelName: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::UpdateChannelName: " << e.what();
exit(EXIT_FAILURE);
}
} }
bool FairMQChannel::IsValid() const bool FairMQChannel::IsValid() const
{ try {
try lock_guard<mutex> lock(fChannelMutex);
{ return fIsValid;
unique_lock<mutex> lock(fChannelMutex); } catch (exception& e) {
return fIsValid; LOG(error) << "Exception caught in FairMQChannel::IsValid: " << e.what();
} throw ChannelConfigurationError(fair::mq::tools::ToString("failed to acquire lock: ", e.what()));
catch (exception& e)
{
LOG(error) << "Exception caught in FairMQChannel::IsValid: " << e.what();
exit(EXIT_FAILURE);
}
} }
bool FairMQChannel::ValidateChannel() bool FairMQChannel::ValidateChannel()
{ try {
try lock_guard<mutex> lock(fChannelMutex);
stringstream ss;
ss << "Validating channel '" << fName << "'... ";
if (fIsValid)
{ {
unique_lock<mutex> lock(fChannelMutex); ss << "ALREADY VALID";
stringstream ss;
ss << "Validating channel \"" << fName << "\"... ";
if (fIsValid)
{
ss << "ALREADY VALID";
LOG(debug) << ss.str();
return true;
}
// validate socket type
const string socketTypeNames[] = { "sub", "pub", "pull", "push", "req", "rep", "xsub", "xpub", "dealer", "router", "pair" };
const set<string> socketTypes(socketTypeNames, socketTypeNames + sizeof(socketTypeNames) / sizeof(string));
if (socketTypes.find(fType) == socketTypes.end())
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "Invalid channel type: \"" << fType << "\"";
exit(EXIT_FAILURE);
}
// validate socket address
if (fAddress == "unspecified" || fAddress == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(debug) << "invalid channel address: \"" << fAddress << "\"";
return false;
}
else
{
vector<string> endpoints;
boost::algorithm::split(endpoints, fAddress, boost::algorithm::is_any_of(";"));
for (const auto endpoint : endpoints)
{
string address;
if (endpoint[0] == '@' || endpoint[0] == '+' || endpoint[0] == '>')
{
address = endpoint.substr(1);
}
else
{
// we don't have a method modifier, check if the default method is set
const string socketMethodNames[] = { "bind", "connect" };
const set<string> socketMethods(socketMethodNames, socketMethodNames + sizeof(socketMethodNames) / sizeof(string));
if (socketMethods.find(fMethod) == socketMethods.end())
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "Invalid endpoint connection method: \"" << fMethod << "\" for " << endpoint;
exit(EXIT_FAILURE);
}
address = endpoint;
}
// check if address is a tcp or ipc address
if (address.compare(0, 6, "tcp://") == 0)
{
// check if TCP address contains port delimiter
string addressString = address.substr(6);
if (addressString.find(':') == string::npos)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: \"" << address << "\" (missing port?)";
return false;
}
}
else if (address.compare(0, 6, "ipc://") == 0)
{
// check if IPC address is not empty
string addressString = address.substr(6);
if (addressString == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: \"" << address << "\" (empty IPC address?)";
return false;
}
}
else if (address.compare(0, 9, "inproc://") == 0)
{
// check if IPC address is not empty
string addressString = address.substr(9);
if (addressString == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: \"" << address << "\" (empty inproc address?)";
return false;
}
}
else if (address.compare(0, 8, "verbs://") == 0)
{
// check if IPC address is not empty
string addressString = address.substr(9);
if (addressString == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: \"" << address << "\" (empty verbs address?)";
return false;
}
}
else
{
// if neither TCP or IPC is specified, return invalid
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: \"" << address << "\" (missing protocol specifier?)";
return false;
}
}
}
// validate socket buffer size for sending
if (fSndBufSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel send buffer size (cannot be negative): \"" << fSndBufSize << "\"";
exit(EXIT_FAILURE);
}
// validate socket buffer size for receiving
if (fRcvBufSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel receive buffer size (cannot be negative): \"" << fRcvBufSize << "\"";
exit(EXIT_FAILURE);
}
// validate socket kernel transmit size for sending
if (fSndKernelSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel send kernel transmit size (cannot be negative): \"" << fSndKernelSize << "\"";
exit(EXIT_FAILURE);
}
// validate socket kernel transmit size for receiving
if (fRcvKernelSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel receive kernel transmit size (cannot be negative): \"" << fRcvKernelSize << "\"";
exit(EXIT_FAILURE);
}
// validate socket rate logging interval
if (fRateLogging < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid socket rate logging interval (cannot be negative): \"" << fRateLogging << "\"";
exit(EXIT_FAILURE);
}
fIsValid = true;
ss << "VALID";
LOG(debug) << ss.str(); LOG(debug) << ss.str();
return true; return true;
} }
catch (exception& e)
// validate socket type
const string socketTypeNames[] = { "sub", "pub", "pull", "push", "req", "rep", "xsub", "xpub", "dealer", "router", "pair" };
const set<string> socketTypes(socketTypeNames, socketTypeNames + sizeof(socketTypeNames) / sizeof(string));
if (socketTypes.find(fType) == socketTypes.end())
{ {
LOG(error) << "Exception caught in FairMQChannel::ValidateChannel: " << e.what(); ss << "INVALID";
exit(EXIT_FAILURE); LOG(debug) << ss.str();
LOG(error) << "Invalid channel type: '" << fType << "'";
throw ChannelConfigurationError(fair::mq::tools::ToString("Invalid channel type: '", fType, "'"));
} }
// validate socket address
if (fAddress == "unspecified" || fAddress == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(debug) << "invalid channel address: '" << fAddress << "'";
return false;
}
else
{
vector<string> endpoints;
boost::algorithm::split(endpoints, fAddress, boost::algorithm::is_any_of(";"));
for (const auto endpoint : endpoints)
{
string address;
if (endpoint[0] == '@' || endpoint[0] == '+' || endpoint[0] == '>')
{
address = endpoint.substr(1);
}
else
{
// we don't have a method modifier, check if the default method is set
const string socketMethodNames[] = { "bind", "connect" };
const set<string> socketMethods(socketMethodNames, socketMethodNames + sizeof(socketMethodNames) / sizeof(string));
if (socketMethods.find(fMethod) == socketMethods.end())
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "Invalid endpoint connection method: '" << fMethod << "' for " << endpoint;
throw ChannelConfigurationError(fair::mq::tools::ToString("Invalid endpoint connection method: '", fMethod, "' for ", endpoint));
}
address = endpoint;
}
// check if address is a tcp or ipc address
if (address.compare(0, 6, "tcp://") == 0)
{
// check if TCP address contains port delimiter
string addressString = address.substr(6);
if (addressString.find(':') == string::npos)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: '" << address << "' (missing port?)";
return false;
}
}
else if (address.compare(0, 6, "ipc://") == 0)
{
// check if IPC address is not empty
string addressString = address.substr(6);
if (addressString == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: '" << address << "' (empty IPC address?)";
return false;
}
}
else if (address.compare(0, 9, "inproc://") == 0)
{
// check if IPC address is not empty
string addressString = address.substr(9);
if (addressString == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: '" << address << "' (empty inproc address?)";
return false;
}
}
else if (address.compare(0, 8, "verbs://") == 0)
{
// check if IPC address is not empty
string addressString = address.substr(9);
if (addressString == "")
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: '" << address << "' (empty verbs address?)";
return false;
}
}
else
{
// if neither TCP or IPC is specified, return invalid
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel address: '" << address << "' (missing protocol specifier?)";
return false;
}
}
}
// validate socket buffer size for sending
if (fSndBufSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel send buffer size (cannot be negative): '" << fSndBufSize << "'";
throw ChannelConfigurationError(fair::mq::tools::ToString("invalid channel send buffer size (cannot be negative): '", fSndBufSize, "'"));
}
// validate socket buffer size for receiving
if (fRcvBufSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel receive buffer size (cannot be negative): '" << fRcvBufSize << "'";
throw ChannelConfigurationError(fair::mq::tools::ToString("invalid channel receive buffer size (cannot be negative): '", fRcvBufSize, "'"));
}
// validate socket kernel transmit size for sending
if (fSndKernelSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel send kernel transmit size (cannot be negative): '" << fSndKernelSize << "'";
throw ChannelConfigurationError(fair::mq::tools::ToString("invalid channel send kernel transmit size (cannot be negative): '", fSndKernelSize, "'"));
}
// validate socket kernel transmit size for receiving
if (fRcvKernelSize < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid channel receive kernel transmit size (cannot be negative): '" << fRcvKernelSize << "'";
throw ChannelConfigurationError(fair::mq::tools::ToString("invalid channel receive kernel transmit size (cannot be negative): '", fRcvKernelSize, "'"));
}
// validate socket rate logging interval
if (fRateLogging < 0)
{
ss << "INVALID";
LOG(debug) << ss.str();
LOG(error) << "invalid socket rate logging interval (cannot be negative): '" << fRateLogging << "'";
throw ChannelConfigurationError(fair::mq::tools::ToString("invalid socket rate logging interval (cannot be negative): '", fRateLogging, "'"));
}
fIsValid = true;
ss << "VALID";
LOG(debug) << ss.str();
return true;
} catch (exception& e) {
LOG(error) << "Exception caught in FairMQChannel::ValidateChannel: " << e.what();
throw ChannelConfigurationError(fair::mq::tools::ToString(e.what()));
} }
void FairMQChannel::InitTransport(shared_ptr<FairMQTransportFactory> factory) void FairMQChannel::InitTransport(shared_ptr<FairMQTransportFactory> factory)
@ -684,7 +564,7 @@ void FairMQChannel::InitTransport(shared_ptr<FairMQTransportFactory> factory)
void FairMQChannel::ResetChannel() void FairMQChannel::ResetChannel()
{ {
unique_lock<mutex> lock(fChannelMutex); lock_guard<mutex> lock(fChannelMutex);
fIsValid = false; fIsValid = false;
// TODO: implement channel resetting // TODO: implement channel resetting
} }

3
fairmq/FairMQChannel.h
View File

@ -14,6 +14,7 @@
#include <vector> #include <vector>
#include <atomic> #include <atomic>
#include <mutex> #include <mutex>
#include <stdexcept>
#include <FairMQTransportFactory.h> #include <FairMQTransportFactory.h>
#include <FairMQSocket.h> #include <FairMQSocket.h>
@ -51,6 +52,8 @@ class FairMQChannel
/// Default destructor /// Default destructor
virtual ~FairMQChannel(); virtual ~FairMQChannel();
struct ChannelConfigurationError : std::runtime_error { using std::runtime_error::runtime_error; };
FairMQSocket& GetSocket() const; FairMQSocket& GetSocket() const;
auto Bind(const std::string& address) -> bool auto Bind(const std::string& address) -> bool

85
fairmq/FairMQDevice.cxx
View File

@ -16,7 +16,6 @@
#include <list> #include <list>
#include <cstdlib> #include <cstdlib>
#include <stdexcept>
#include <random> #include <random>
#include <chrono> #include <chrono>
#include <mutex> #include <mutex>
@ -56,9 +55,6 @@ FairMQDevice::FairMQDevice(FairMQProgOptions* config, const fair::mq::tools::Ver
, fInternalConfig(config ? nullptr : fair::mq::tools::make_unique<FairMQProgOptions>()) , fInternalConfig(config ? nullptr : fair::mq::tools::make_unique<FairMQProgOptions>())
, fConfig(config ? config : fInternalConfig.get()) , fConfig(config ? config : fInternalConfig.get())
, fId() , fId()
, fInitialValidationFinished(false)
, fInitialValidationCondition()
, fInitialValidationMutex()
, fPortRangeMin(22000) , fPortRangeMin(22000)
, fPortRangeMax(32000) , fPortRangeMax(32000)
, fDefaultTransportType(fair::mq::Transport::DEFAULT) , fDefaultTransportType(fair::mq::Transport::DEFAULT)
@ -76,6 +72,7 @@ FairMQDevice::FairMQDevice(FairMQProgOptions* config, const fair::mq::tools::Ver
, fInterrupted(false) , fInterrupted(false)
, fInterruptedCV() , fInterruptedCV()
, fInterruptedMtx() , fInterruptedMtx()
, fRateLogging(true)
{ {
} }
@ -86,13 +83,12 @@ void FairMQDevice::InitWrapper()
fPortRangeMin = fConfig->GetValue<int>("port-range-min"); fPortRangeMin = fConfig->GetValue<int>("port-range-min");
fPortRangeMax = fConfig->GetValue<int>("port-range-max"); fPortRangeMax = fConfig->GetValue<int>("port-range-max");
try try {
{
fDefaultTransportType = fair::mq::TransportTypes.at(fConfig->GetValue<string>("transport")); fDefaultTransportType = fair::mq::TransportTypes.at(fConfig->GetValue<string>("transport"));
} } catch (const exception& e) {
catch (const exception& e) LOG(error) << "exception: " << e.what();
{
LOG(error) << "invalid transport type provided: " << fConfig->GetValue<string>("transport"); LOG(error) << "invalid transport type provided: " << fConfig->GetValue<string>("transport");
throw;
} }
for (auto& c : fConfig->GetFairMQMap()) for (auto& c : fConfig->GetFairMQMap())
@ -159,9 +155,8 @@ void FairMQDevice::InitWrapper()
} }
else else
{ {
LOG(error) << "Cannot update configuration. Socket method (bind/connect) not specified."; LOG(error) << "Cannot update configuration. Socket method (bind/connect) for channel '" << vi->fName << "' not specified.";
ChangeState(ERROR_FOUND); throw runtime_error(fair::mq::tools::ToString("Cannot update configuration. Socket method (bind/connect) for channel ", vi->fName, " not specified."));
// throw runtime_error("Cannot update configuration. Socket method (bind/connect) not specified.");
} }
// } // }
} }
@ -174,19 +169,11 @@ void FairMQDevice::InitWrapper()
if (!uninitializedBindingChannels.empty()) if (!uninitializedBindingChannels.empty())
{ {
LOG(error) << uninitializedBindingChannels.size() << " of the binding channels could not initialize. Initial configuration incomplete."; LOG(error) << uninitializedBindingChannels.size() << " of the binding channels could not initialize. Initial configuration incomplete.";
ChangeState(ERROR_FOUND); throw runtime_error(fair::mq::tools::ToString(uninitializedBindingChannels.size(), " of the binding channels could not initialize. Initial configuration incomplete."));
// throw runtime_error(fair::mq::tools::ToString(uninitializedBindingChannels.size(), " of the binding channels could not initialize. Initial configuration incomplete."));
} }
CallStateChangeCallbacks(INITIALIZING_DEVICE); CallStateChangeCallbacks(INITIALIZING_DEVICE);
// notify parent thread about completion of first validation.
{
lock_guard<mutex> lock(fInitialValidationMutex);
fInitialValidationFinished = true;
fInitialValidationCondition.notify_one();
}
int initializationTimeoutInS = fConfig->GetValue<int>("initialization-timeout"); int initializationTimeoutInS = fConfig->GetValue<int>("initialization-timeout");
// go over the list of channels until all are initialized (and removed from the uninitialized list) // go over the list of channels until all are initialized (and removed from the uninitialized list)
@ -200,24 +187,20 @@ void FairMQDevice::InitWrapper()
{ {
this_thread::sleep_for(chrono::milliseconds(sleepTimeInMS)); this_thread::sleep_for(chrono::milliseconds(sleepTimeInMS));
if (fConfig != nullptr) for (auto& chan : uninitializedConnectingChannels)
{ {
for (auto& chan : uninitializedConnectingChannels) string key{"chans." + chan->GetChannelPrefix() + "." + chan->GetChannelIndex() + ".address"};
string newAddress = fConfig->GetValue<string>(key);
if (newAddress != chan->GetAddress())
{ {
string key{"chans." + chan->GetChannelPrefix() + "." + chan->GetChannelIndex() + ".address"}; chan->UpdateAddress(newAddress);
string newAddress = fConfig->GetValue<string>(key);
if (newAddress != chan->GetAddress())
{
chan->UpdateAddress(newAddress);
}
} }
} }
if (numAttempts++ > maxAttempts) if (numAttempts++ > maxAttempts)
{ {
LOG(error) << "could not connect all channels after " << initializationTimeoutInS << " attempts"; LOG(error) << "could not connect all channels after " << initializationTimeoutInS << " attempts";
ChangeState(ERROR_FOUND); throw runtime_error(fair::mq::tools::ToString("could not connect all channels after ", initializationTimeoutInS, " attempts"));
// throw runtime_error(fair::mq::tools::ToString("could not connect all channels after ", initializationTimeoutInS, " attempts"));
} }
AttachChannels(uninitializedConnectingChannels); AttachChannels(uninitializedConnectingChannels);
@ -225,13 +208,11 @@ void FairMQDevice::InitWrapper()
Init(); Init();
ChangeState(internal_DEVICE_READY); if (fChannels.empty()) {
} LOG(warn) << "No channels created after finishing initialization";
}
void FairMQDevice::WaitForInitialValidation() ChangeState(internal_DEVICE_READY);
{
unique_lock<mutex> lock(fInitialValidationMutex);
fInitialValidationCondition.wait(lock, [&] () { return fInitialValidationFinished; });
} }
void FairMQDevice::Init() void FairMQDevice::Init()
@ -382,11 +363,8 @@ bool FairMQDevice::AttachChannel(FairMQChannel& ch)
if (newAddress != ch.fAddress) if (newAddress != ch.fAddress)
{ {
ch.UpdateAddress(newAddress); ch.UpdateAddress(newAddress);
if (fConfig) string key{"chans." + ch.GetChannelPrefix() + "." + ch.GetChannelIndex() + ".address"};
{ fConfig->SetValue(key, newAddress);
string key{"chans." + ch.GetChannelPrefix() + "." + ch.GetChannelIndex() + ".address"};
fConfig->SetValue(key, newAddress);
}
} }
return true; return true;
@ -495,6 +473,7 @@ void FairMQDevice::RunWrapper()
LOG(info) << "DEVICE: Running..."; LOG(info) << "DEVICE: Running...";
// start the rate logger thread // start the rate logger thread
fRateLogging = true;
future<void> rateLogger = async(launch::async, &FairMQDevice::LogSocketRates, this); future<void> rateLogger = async(launch::async, &FairMQDevice::LogSocketRates, this);
// notify transports to resume transfers // notify transports to resume transfers
@ -507,8 +486,7 @@ void FairMQDevice::RunWrapper()
t.second->Resume(); t.second->Resume();
} }
try try {
{
PreRun(); PreRun();
// process either data callbacks or ConditionalRun/Run // process either data callbacks or ConditionalRun/Run
@ -538,11 +516,14 @@ void FairMQDevice::RunWrapper()
Run(); Run();
} }
} } catch (const out_of_range& oor) {
catch (const out_of_range& oor)
{
LOG(error) << "out of range: " << oor.what(); LOG(error) << "out of range: " << oor.what();
LOG(error) << "incorrect/incomplete channel configuration?"; LOG(error) << "incorrect/incomplete channel configuration?";
fRateLogging = false;
throw;
} catch (...) {
fRateLogging = false;
throw;
} }
// if Run() exited and the state is still RUNNING, transition to READY. // if Run() exited and the state is still RUNNING, transition to READY.
@ -721,7 +702,7 @@ void FairMQDevice::PollForTransport(const FairMQTransportFactory* factory, const
catch (exception& e) catch (exception& e)
{ {
LOG(error) << "FairMQDevice::PollForTransport() failed: " << e.what() << ", going to ERROR state."; LOG(error) << "FairMQDevice::PollForTransport() failed: " << e.what() << ", going to ERROR state.";
ChangeState(ERROR_FOUND); throw runtime_error(fair::mq::tools::ToString("FairMQDevice::PollForTransport() failed: ", e.what(), ", going to ERROR state."));
} }
} }
@ -877,7 +858,7 @@ void FairMQDevice::LogSocketRates()
LOG(debug) << "<channel>: in: <#msgs> (<MB>) out: <#msgs> (<MB>)"; LOG(debug) << "<channel>: in: <#msgs> (<MB>) out: <#msgs> (<MB>)";
while (CheckCurrentState(RUNNING)) while (fRateLogging)
{ {
t1 = chrono::high_resolution_clock::now(); t1 = chrono::high_resolution_clock::now();
@ -931,6 +912,7 @@ void FairMQDevice::Unblock()
{ {
lock_guard<mutex> guard(fInterruptedMtx); lock_guard<mutex> guard(fInterruptedMtx);
fInterrupted = true; fInterrupted = true;
fRateLogging = false;
} }
fInterruptedCV.notify_all(); fInterruptedCV.notify_all();
} }
@ -977,11 +959,6 @@ void FairMQDevice::Reset()
{ {
} }
const FairMQChannel& FairMQDevice::GetChannel(const string& channelName, const int index) const
{
return fChannels.at(channelName).at(index);
}
void FairMQDevice::Exit() void FairMQDevice::Exit()
{ {
} }

71
fairmq/FairMQDevice.h
View File

@ -31,6 +31,7 @@
#include <functional> #include <functional>
#include <assert.h> // static_assert #include <assert.h> // static_assert
#include <type_traits> // is_trivially_copyable #include <type_traits> // is_trivially_copyable
#include <stdexcept>
#include <mutex> #include <mutex>
#include <condition_variable> #include <condition_variable>
@ -102,9 +103,9 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index /// @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) /// @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. -2 If queueing was not possible or timed out. -1 if there was an error. /// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out. -1 if there was an error.
int Send(FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1) int Send(FairMQMessagePtr& msg, const std::string& channel, const int index = 0, int sndTimeoutInMs = -1)
{ {
return fChannels.at(chan).at(i).Send(msg, sndTimeoutInMs); return GetChannel(channel, index).Send(msg, sndTimeoutInMs);
} }
/// Shorthand method to receive `msg` on `chan` at index `i` /// Shorthand method to receive `msg` on `chan` at index `i`
@ -113,18 +114,18 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index /// @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) /// @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. -2 if reading from the queue was not possible or timed out. -1 if there was an error. /// @return Number of bytes that have been received. -2 if reading from the queue was not possible or timed out. -1 if there was an error.
int Receive(FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1) int Receive(FairMQMessagePtr& msg, const std::string& channel, const int index = 0, int rcvTimeoutInMs = -1)
{ {
return fChannels.at(chan).at(i).Receive(msg, rcvTimeoutInMs); return GetChannel(channel, index).Receive(msg, rcvTimeoutInMs);
} }
int SendAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, \"channelA\", subchannelIndex, timeout);"))) int SendAsync(FairMQMessagePtr& msg, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, \"channelA\", subchannelIndex, timeout);")))
{ {
return fChannels.at(chan).at(i).Send(msg, 0); return GetChannel(channel, index).Send(msg, 0);
} }
int ReceiveAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, \"channelA\", subchannelIndex, timeout);"))) int ReceiveAsync(FairMQMessagePtr& msg, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, \"channelA\", subchannelIndex, timeout);")))
{ {
return fChannels.at(chan).at(i).Receive(msg, 0); return GetChannel(channel, index).Receive(msg, 0);
} }
/// Shorthand method to send FairMQParts on `chan` at index `i` /// Shorthand method to send FairMQParts on `chan` at index `i`
@ -133,9 +134,9 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index /// @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) /// @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. -2 If queueing was not possible or timed out. -1 if there was an error. /// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out. -1 if there was an error.
int64_t Send(FairMQParts& parts, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1) int64_t Send(FairMQParts& parts, const std::string& channel, const int index = 0, int sndTimeoutInMs = -1)
{ {
return fChannels.at(chan).at(i).Send(parts.fParts, sndTimeoutInMs); return GetChannel(channel, index).Send(parts.fParts, sndTimeoutInMs);
} }
/// Shorthand method to receive FairMQParts on `chan` at index `i` /// Shorthand method to receive FairMQParts on `chan` at index `i`
@ -144,18 +145,18 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index /// @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) /// @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. -2 if reading from the queue was not possible or timed out. -1 if there was an error. /// @return Number of bytes that have been received. -2 if reading from the queue was not possible or timed out. -1 if there was an error.
int64_t Receive(FairMQParts& parts, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1) int64_t Receive(FairMQParts& parts, const std::string& channel, const int index = 0, int rcvTimeoutInMs = -1)
{ {
return fChannels.at(chan).at(i).Receive(parts.fParts, rcvTimeoutInMs); return GetChannel(channel, index).Receive(parts.fParts, rcvTimeoutInMs);
} }
int64_t SendAsync(FairMQParts& parts, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, \"channelA\", subchannelIndex, timeout);"))) int64_t SendAsync(FairMQParts& parts, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, \"channelA\", subchannelIndex, timeout);")))
{ {
return fChannels.at(chan).at(i).Send(parts.fParts, 0); return GetChannel(channel, index).Send(parts.fParts, 0);
} }
int64_t ReceiveAsync(FairMQParts& parts, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, \"channelA\", subchannelIndex, timeout);"))) int64_t ReceiveAsync(FairMQParts& parts, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, \"channelA\", subchannelIndex, timeout);")))
{ {
return fChannels.at(chan).at(i).Receive(parts.fParts, 0); return GetChannel(channel, index).Receive(parts.fParts, 0);
} }
/// @brief Getter for default transport factory /// @brief Getter for default transport factory
@ -173,7 +174,7 @@ class FairMQDevice : public FairMQStateMachine
template<typename... Args> template<typename... Args>
FairMQMessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args) FairMQMessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args)
{ {
return fChannels.at(channel).at(index).NewMessage(std::forward<Args>(args)...); return GetChannel(channel, index).NewMessage(std::forward<Args>(args)...);
} }
template<typename T> template<typename T>
@ -185,7 +186,7 @@ class FairMQDevice : public FairMQStateMachine
template<typename T> template<typename T>
FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data) FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data)
{ {
return fChannels.at(channel).at(index).NewStaticMessage(data); return GetChannel(channel, index).NewStaticMessage(data);
} }
template<typename T> template<typename T>
@ -197,7 +198,7 @@ class FairMQDevice : public FairMQStateMachine
template<typename T> template<typename T>
FairMQMessagePtr NewSimpleMessageFor(const std::string& channel, int index, const T& data) FairMQMessagePtr NewSimpleMessageFor(const std::string& channel, int index, const T& data)
{ {
return fChannels.at(channel).at(index).NewSimpleMessage(data); return GetChannel(channel, index).NewSimpleMessage(data);
} }
FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size) FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size)
@ -207,7 +208,7 @@ class FairMQDevice : public FairMQStateMachine
FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size, FairMQRegionCallback callback = nullptr) FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size, FairMQRegionCallback callback = nullptr)
{ {
return fChannels.at(channel).at(index).Transport()->CreateUnmanagedRegion(size, callback); return GetChannel(channel, index).Transport()->CreateUnmanagedRegion(size, callback);
} }
template<typename ...Ts> template<typename ...Ts>
@ -218,19 +219,19 @@ class FairMQDevice : public FairMQStateMachine
// if more than one channel provided, check compatibility // if more than one channel provided, check compatibility
if (chans.size() > 1) if (chans.size() > 1)
{ {
fair::mq::Transport type = fChannels.at(chans.at(0)).at(0).Transport()->GetType(); fair::mq::Transport type = GetChannel(chans.at(0), 0).Transport()->GetType();
for (unsigned int i = 1; i < chans.size(); ++i) for (unsigned int i = 1; i < chans.size(); ++i)
{ {
if (type != fChannels.at(chans.at(i)).at(0).Transport()->GetType()) if (type != GetChannel(chans.at(i), 0).Transport()->GetType())
{ {
LOG(error) << "poller failed: different transports within same poller are not yet supported. Going to ERROR state."; LOG(error) << "poller failed: different transports within same poller are not yet supported. Going to ERROR state.";
ChangeState(ERROR_FOUND); throw std::runtime_error("poller failed: different transports within same poller are not yet supported.");
} }
} }
} }
return fChannels.at(chans.at(0)).at(0).Transport()->CreatePoller(fChannels, chans); return GetChannel(chans.at(0), 0).Transport()->CreatePoller(fChannels, chans);
} }
FairMQPollerPtr NewPoller(const std::vector<FairMQChannel*>& channels) FairMQPollerPtr NewPoller(const std::vector<FairMQChannel*>& channels)
@ -245,7 +246,7 @@ class FairMQDevice : public FairMQStateMachine
if (type != channels.at(i)->Transport()->GetType()) if (type != channels.at(i)->Transport()->GetType())
{ {
LOG(error) << "poller failed: different transports within same poller are not yet supported. Going to ERROR state."; LOG(error) << "poller failed: different transports within same poller are not yet supported. Going to ERROR state.";
ChangeState(ERROR_FOUND); throw std::runtime_error("poller failed: different transports within same poller are not yet supported.");
} }
} }
} }
@ -254,7 +255,7 @@ class FairMQDevice : public FairMQStateMachine
} }
/// Waits for the first initialization run to finish /// Waits for the first initialization run to finish
void WaitForInitialValidation(); void WaitForInitialValidation() __attribute__((deprecated("This method will have no effect in future versions and will be removed. Instead subscribe for state changes and inspect configuration values."))) {}
/// Adds a transport to the device if it doesn't exist /// Adds a transport to the device if it doesn't exist
/// @param transport Transport string ("zeromq"/"nanomsg"/"shmem") /// @param transport Transport string ("zeromq"/"nanomsg"/"shmem")
@ -273,6 +274,7 @@ class FairMQDevice : public FairMQStateMachine
/// @param rhs Left hand side value for comparison /// @param rhs Left hand side value for comparison
static bool SortSocketsByAddress(const FairMQChannel &lhs, const FairMQChannel &rhs); static bool SortSocketsByAddress(const FairMQChannel &lhs, const FairMQChannel &rhs);
// overload to easily bind member functions
template<typename T> template<typename T>
void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQMessagePtr& msg, int index)) void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQMessagePtr& msg, int index))
{ {
@ -299,6 +301,7 @@ class FairMQDevice : public FairMQStateMachine
} }
} }
// overload to easily bind member functions
template<typename T> template<typename T>
void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQParts& parts, int index)) void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQParts& parts, int index))
{ {
@ -325,7 +328,15 @@ class FairMQDevice : public FairMQStateMachine
} }
} }
const FairMQChannel& GetChannel(const std::string& channelName, const int index = 0) const; FairMQChannel& GetChannel(const std::string& channelName, const int index = 0)
try {
return fChannels.at(channelName).at(index);
} catch (const std::out_of_range& oor) {
LOG(error) << "out of range: " << oor.what();
LOG(error) << "requested channel has not been configured? check channel names/configuration.";
fRateLogging = false;
throw;
}
virtual void RegisterChannelEndpoints() {} virtual void RegisterChannelEndpoints() {}
@ -443,11 +454,6 @@ class FairMQDevice : public FairMQStateMachine
virtual void Reset(); virtual void Reset();
private: private:
// condition variable to notify parent thread about end of initial validation.
bool fInitialValidationFinished;
std::condition_variable fInitialValidationCondition;
std::mutex fInitialValidationMutex;
int fPortRangeMin; ///< Minimum value for the port range (if dynamic) int fPortRangeMin; ///< Minimum value for the port range (if dynamic)
int fPortRangeMax; ///< Maximum value for the port range (if dynamic) int fPortRangeMax; ///< Maximum value for the port range (if dynamic)
@ -511,6 +517,7 @@ class FairMQDevice : public FairMQStateMachine
std::atomic<bool> fInterrupted; std::atomic<bool> fInterrupted;
std::condition_variable fInterruptedCV; std::condition_variable fInterruptedCV;
std::mutex fInterruptedMtx; std::mutex fInterruptedMtx;
mutable std::atomic<bool> fRateLogging;
}; };
#endif /* FAIRMQDEVICE_H_ */ #endif /* FAIRMQDEVICE_H_ */