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

convert log severities use to lowercase and remove use of MQLOG

Browse Source
This commit is contained in:
Alexey Rybalchenko
2017-12-22 10:40:50 +01:00
committed by Mohammad Al-Turany
parent 4e942e489b
commit a3393e600e
58 changed files with 607 additions and 609 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
fairmq/shmem/FairMQMessageSHM.cxx
View File

@@ -39,7 +39,7 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager)
{
if (zmq_msg_init(&fMessage) != 0)
{
LOG(ERROR) << "failed initializing message, reason: " << zmq_strerror(errno);
LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);
}
fMetaCreated = true;
}
@@ -104,7 +104,7 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQUnmanagedRegionPtr& r
if (zmq_msg_init_size(&fMessage, sizeof(MetaHeader)) != 0)
{
LOG(ERROR) << "failed initializing meta message, reason: " << zmq_strerror(errno);
LOG(error) << "failed initializing meta message, reason: " << zmq_strerror(errno);
}
else
{
@@ -120,8 +120,8 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQUnmanagedRegionPtr& r
}
else
{
LOG(ERROR) << "shmem: trying to create region message with data from outside the region";
throw runtime_error("shmem: trying to create region message with data from outside the region");
LOG(error) << "trying to create region message with data from outside the region";
throw runtime_error("trying to create region message with data from outside the region");
}
}
@@ -137,7 +137,7 @@ bool FairMQMessageSHM::InitializeChunk(const size_t size)
}
catch (bipc::bad_alloc& ba)
{
// LOG(WARN) << "Shared memory full...";
// LOG(warn) << "Shared memory full...";
this_thread::sleep_for(chrono::milliseconds(50));
if (fInterrupted)
{
@@ -155,7 +155,7 @@ bool FairMQMessageSHM::InitializeChunk(const size_t size)
if (zmq_msg_init_size(&fMessage, sizeof(MetaHeader)) != 0)
{
LOG(ERROR) << "failed initializing meta message, reason: " << zmq_strerror(errno);
LOG(error) << "failed initializing meta message, reason: " << zmq_strerror(errno);
return false;
}
MetaHeader header;
@@ -178,7 +178,7 @@ void FairMQMessageSHM::Rebuild()
if (zmq_msg_init(&fMessage) != 0)
{
LOG(ERROR) << "failed initializing message, reason: " << zmq_strerror(errno);
LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);
}
fMetaCreated = true;
}
@@ -238,7 +238,7 @@ void* FairMQMessageSHM::GetData() const
}
else
{
// LOG(WARN) << "could not get pointer from a region message";
// LOG(warn) << "could not get pointer from a region message";
fLocalPtr = nullptr;
}
return fLocalPtr;
@@ -273,13 +273,13 @@ bool FairMQMessageSHM::SetUsedSize(const size_t size)
}
catch (bipc::interprocess_exception& e)
{
LOG(INFO) << "FairMQMessageSHM::SetUsedSize could not set used size: " << e.what();
LOG(info) << "could not set used size: " << e.what();
return false;
}
}
else
{
LOG(ERROR) << "FairMQMessageSHM::SetUsedSize: cannot set used size higher than original.";
LOG(error) << "cannot set used size higher than original.";
return false;
}
}
@@ -303,12 +303,12 @@ void FairMQMessageSHM::Copy(const FairMQMessage& msg)
}
else
{
LOG(ERROR) << "FairMQMessageSHM::Copy() fail: source message not initialized!";
LOG(error) << "copy fail: source message not initialized!";
}
}
else
{
LOG(ERROR) << "FairMQMessageSHM::Copy() fail: target message already initialized!";
LOG(error) << "copy fail: target message already initialized!";
}
}
@@ -326,12 +326,12 @@ void FairMQMessageSHM::Copy(const FairMQMessagePtr& msg)
}
else
{
LOG(ERROR) << "FairMQMessageSHM::Copy() fail: source message not initialized!";
LOG(error) << "copy fail: source message not initialized!";
}
}
else
{
LOG(ERROR) << "FairMQMessageSHM::Copy() fail: target message already initialized!";
LOG(error) << "copy fail: target message already initialized!";
}
}
@@ -350,11 +350,11 @@ void FairMQMessageSHM::CloseMessage()
// RegionBlock block(fHandle, fSize);
// if (fManager.GetRegionQueue(fRegionId).try_send(static_cast<void*>(&block), sizeof(RegionBlock), 0))
// {
// // LOG(INFO) << "true";
// // LOG(info) << "true";
// }
// // else
// // {
// // LOG(DEBUG) << "could not send ack";
// // LOG(debug) << "could not send ack";
// // }
// timed version
@@ -379,12 +379,12 @@ void FairMQMessageSHM::CloseMessage()
{
break;
}
LOG(DEBUG) << "region ack queue is full, retrying...";
LOG(debug) << "region ack queue is full, retrying...";
}
}
else
{
// LOG(WARN) << "region ack queue for id " << fRegionId << " no longer exist. Not sending ack";
// LOG(warn) << "region ack queue for id " << fRegionId << " no longer exist. Not sending ack";
success = true;
}
}
@@ -396,7 +396,7 @@ void FairMQMessageSHM::CloseMessage()
{
if (zmq_msg_close(&fMessage) != 0)
{
LOG(ERROR) << "failed closing message, reason: " << zmq_strerror(errno);
LOG(error) << "failed closing message, reason: " << zmq_strerror(errno);
}
}
}

20
fairmq/shmem/FairMQPollerSHM.cxx
View File

@@ -103,8 +103,8 @@ FairMQPollerSHM::FairMQPollerSHM(const unordered_map<string, vector<FairMQChanne
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "shmem: at least one of the provided channel keys for poller initialization is invalid";
LOG(ERROR) << "shmem: out of range error: " << oor.what() << '\n';
LOG(error) << "at least one of the provided channel keys for poller initialization is invalid";
LOG(error) << "out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}
@@ -148,7 +148,7 @@ void FairMQPollerSHM::SetItemEvents(zmq_pollitem_t& item, const int type)
}
else
{
LOG(ERROR) << "zeromq: invalid poller configuration, exiting.";
LOG(error) << "invalid poller configuration, exiting.";
exit(EXIT_FAILURE);
}
}
@@ -159,12 +159,12 @@ void FairMQPollerSHM::Poll(const int timeout)
{
if (errno == ETERM)
{
LOG(DEBUG) << "shmem: polling exited, reason: " << zmq_strerror(errno);
LOG(debug) << "polling exited, reason: " << zmq_strerror(errno);
}
else
{
LOG(ERROR) << "shmem: polling failed, reason: " << zmq_strerror(errno);
throw std::runtime_error("shmem: polling failed");
LOG(error) << "polling failed, reason: " << zmq_strerror(errno);
throw std::runtime_error("polling failed");
}
}
}
@@ -202,8 +202,8 @@ bool FairMQPollerSHM::CheckInput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "shmem: invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "shmem: out of range error: " << oor.what() << '\n';
LOG(error) << "invalid channel key: \"" << channelKey << "\"";
LOG(error) << "out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}
@@ -221,8 +221,8 @@ bool FairMQPollerSHM::CheckOutput(const string channelKey, const int index)
}
catch (const std::out_of_range& oor)
{
LOG(ERROR) << "shmem: Invalid channel key: \"" << channelKey << "\"";
LOG(ERROR) << "shmem: out of range error: " << oor.what() << '\n';
LOG(error) << "Invalid channel key: \"" << channelKey << "\"";
LOG(error) << "out of range error: " << oor.what() << '\n';
exit(EXIT_FAILURE);
}
}

78
fairmq/shmem/FairMQSocketSHM.cxx
View File

@@ -36,13 +36,13 @@ FairMQSocketSHM::FairMQSocketSHM(Manager& manager, const string& type, const str
if (fSocket == nullptr)
{
LOG(ERROR) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);
exit(EXIT_FAILURE);
}
if (zmq_setsockopt(fSocket, ZMQ_IDENTITY, fId.c_str(), fId.length()) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_IDENTITY socket option, reason: " << zmq_strerror(errno);
LOG(error) << "Failed setting ZMQ_IDENTITY socket option, reason: " << zmq_strerror(errno);
}
// Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating.
@@ -50,30 +50,30 @@ FairMQSocketSHM::FairMQSocketSHM(Manager& manager, const string& type, const str
int linger = 1000;
if (zmq_setsockopt(fSocket, ZMQ_LINGER, &linger, sizeof(linger)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
LOG(error) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
}
int sndTimeout = 700;
if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &sndTimeout, sizeof(sndTimeout)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);
LOG(error) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);
}
int rcvTimeout = 700;
if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &rcvTimeout, sizeof(rcvTimeout)) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);
LOG(error) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);
}
if (type == "sub")
{
if (zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, nullptr, 0) != 0)
{
LOG(ERROR) << "Failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);
LOG(error) << "Failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);
}
}
// LOG(INFO) << "created socket " << fId;
// LOG(info) << "created socket " << fId;
}
string FairMQSocketSHM::GetId()
@@ -83,7 +83,7 @@ string FairMQSocketSHM::GetId()
bool FairMQSocketSHM::Bind(const string& address)
{
// LOG(INFO) << "bind socket " << fId << " on " << address;
// LOG(info) << "bind socket " << fId << " on " << address;
if (zmq_bind(fSocket, address.c_str()) != 0)
{
@@ -91,7 +91,7 @@ bool FairMQSocketSHM::Bind(const string& address)
// do not print error in this case, this is handled by FairMQDevice in case no connection could be established after trying a number of random ports from a range.
return false;
}
LOG(ERROR) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
return true;
@@ -99,11 +99,11 @@ bool FairMQSocketSHM::Bind(const string& address)
void FairMQSocketSHM::Connect(const string& address)
{
// LOG(INFO) << "connect socket " << fId << " on " << address;
// LOG(info) << "connect socket " << fId << " on " << address;
if (zmq_connect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
// error here means incorrect configuration. exit if it happens.
exit(EXIT_FAILURE);
}
@@ -142,12 +142,12 @@ int FairMQSocketSHM::Send(FairMQMessagePtr& msg, const int flags)
}
else if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
LOG(info) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
@@ -196,12 +196,12 @@ int FairMQSocketSHM::Receive(FairMQMessagePtr& msg, const int flags)
}
else if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
LOG(info) << "terminating socket " << fId;
return -1;
}
else
{
LOG(ERROR) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
@@ -249,10 +249,10 @@ int64_t FairMQSocketSHM::Send(vector<FairMQMessagePtr>& msgVec, const int flags)
}
if (zmq_errno() == ETERM)
{
LOG(INFO) << "terminating socket " << fId;
LOG(info) << "terminating socket " << fId;
return -1;
}
LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
return nbytes;
}
}
@@ -276,7 +276,7 @@ int64_t FairMQSocketSHM::Send(vector<FairMQMessagePtr>& msgVec, const int flags)
}
else // if the vector is empty, something might be wrong
{
LOG(WARN) << "Will not send empty vector";
LOG(warn) << "Will not send empty vector";
return -1;
}
}
@@ -292,7 +292,7 @@ int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int fla
// Warn if the vector is filled before Receive() and empty it.
// if (msgVec.size() > 0)
// {
// LOG(WARN) << "Message vector contains elements before Receive(), they will be deleted!";
// LOG(warn) << "Message vector contains elements before Receive(), they will be deleted!";
// msgVec.clear();
// }
@@ -360,7 +360,7 @@ int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int fla
void FairMQSocketSHM::Close()
{
// LOG(DEBUG) << "Closing socket " << fId;
// LOG(debug) << "Closing socket " << fId;
if (fSocket == nullptr)
{
@@ -369,7 +369,7 @@ void FairMQSocketSHM::Close()
if (zmq_close(fSocket) != 0)
{
LOG(ERROR) << "Failed closing socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed closing socket " << fId << ", reason: " << zmq_strerror(errno);
}
fSocket = nullptr;
@@ -404,7 +404,7 @@ void FairMQSocketSHM::SetOption(const string& option, const void* value, size_t
{
if (zmq_setsockopt(fSocket, GetConstant(option), value, valueSize) < 0)
{
LOG(ERROR) << "Failed setting socket option, reason: " << zmq_strerror(errno);
LOG(error) << "Failed setting socket option, reason: " << zmq_strerror(errno);
}
}
@@ -412,7 +412,7 @@ void FairMQSocketSHM::GetOption(const string& option, void* value, size_t* value
{
if (zmq_getsockopt(fSocket, GetConstant(option), value, valueSize) < 0)
{
LOG(ERROR) << "Failed getting socket option, reason: " << zmq_strerror(errno);
LOG(error) << "Failed getting socket option, reason: " << zmq_strerror(errno);
}
}
@@ -442,17 +442,17 @@ bool FairMQSocketSHM::SetSendTimeout(const int timeout, const string& address, c
{
if (zmq_unbind(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed unbinding socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed unbinding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
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);
LOG(error) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
@@ -460,23 +460,23 @@ bool FairMQSocketSHM::SetSendTimeout(const int timeout, const string& address, c
{
if (zmq_disconnect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed disconnecting socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed disconnecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
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);
LOG(error) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
else
{
LOG(ERROR) << "SetSendTimeout() failed - unknown method provided!";
LOG(error) << "timeout failed - unknown method provided!";
return false;
}
@@ -490,7 +490,7 @@ int FairMQSocketSHM::GetSendTimeout() const
if (zmq_getsockopt(fSocket, ZMQ_SNDTIMEO, &timeout, &size) != 0)
{
LOG(ERROR) << "Failed getting option 'receive timeout' on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed getting option 'receive timeout' on socket " << fId << ", reason: " << zmq_strerror(errno);
}
return timeout;
@@ -502,17 +502,17 @@ bool FairMQSocketSHM::SetReceiveTimeout(const int timeout, const string& address
{
if (zmq_unbind(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed unbinding socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed unbinding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
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);
LOG(error) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
@@ -520,23 +520,23 @@ bool FairMQSocketSHM::SetReceiveTimeout(const int timeout, const string& address
{
if (zmq_disconnect(fSocket, address.c_str()) != 0)
{
LOG(ERROR) << "Failed disconnecting socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed disconnecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &timeout, sizeof(int)) != 0)
{
LOG(ERROR) << "Failed setting option on socket " << fId << ", reason: " << zmq_strerror(errno);
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);
LOG(error) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
}
else
{
LOG(ERROR) << "SetReceiveTimeout() failed - unknown method provided!";
LOG(error) << "timeout failed - unknown method provided!";
return false;
}
@@ -550,7 +550,7 @@ int FairMQSocketSHM::GetReceiveTimeout() const
if (zmq_getsockopt(fSocket, ZMQ_RCVTIMEO, &timeout, &size) != 0)
{
LOG(ERROR) << "Failed getting option 'receive timeout' on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed getting option 'receive timeout' on socket " << fId << ", reason: " << zmq_strerror(errno);
}
return timeout;

48
fairmq/shmem/FairMQTransportFactorySHM.cxx
View File

@@ -46,13 +46,13 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
{
int major, minor, patch;
zmq_version(&major, &minor, &patch);
LOG(DEBUG) << "Transport: Using ZeroMQ (" << major << "." << minor << "." << patch << ") & "
LOG(debug) << "Transport: Using ZeroMQ (" << major << "." << minor << "." << patch << ") & "
<< "boost::interprocess (" << (BOOST_VERSION / 100000) << "." << (BOOST_VERSION / 100 % 1000) << "." << (BOOST_VERSION % 100) << ")";
fContext = zmq_ctx_new();
if (!fContext)
{
LOG(ERROR) << "failed creating context, reason: " << zmq_strerror(errno);
LOG(error) << "failed creating context, reason: " << zmq_strerror(errno);
exit(EXIT_FAILURE);
}
@@ -68,7 +68,7 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
}
else
{
LOG(WARN) << "shmem: FairMQProgOptions not available! Using defaults.";
LOG(warn) << "FairMQProgOptions not available! Using defaults.";
}
try
@@ -77,17 +77,17 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
if (zmq_ctx_set(fContext, ZMQ_IO_THREADS, numIoThreads) != 0)
{
LOG(ERROR) << "shmem: failed configuring context, reason: " << zmq_strerror(errno);
LOG(error) << "failed configuring context, reason: " << zmq_strerror(errno);
}
// Set the maximum number of allowed sockets on the context.
if (zmq_ctx_set(fContext, ZMQ_MAX_SOCKETS, 10000) != 0)
{
LOG(ERROR) << "shmem: failed configuring context, reason: " << zmq_strerror(errno);
LOG(error) << "failed configuring context, reason: " << zmq_strerror(errno);
}
fManager = fair::mq::tools::make_unique<Manager>(fSessionName, segmentSize);
LOG(DEBUG) << "shmem: created/opened shared memory segment '" << "fmq_shm_" << fSessionName << "_main" << "' of " << segmentSize << " bytes. Available are " << fManager->Segment().get_free_memory() << " bytes.";
LOG(debug) << "created/opened shared memory segment '" << "fmq_shm_" << fSessionName << "_main" << "' of " << segmentSize << " bytes. Available are " << fManager->Segment().get_free_memory() << " bytes.";
{
bipc::scoped_lock<bipc::named_mutex> lock(*fShMutex);
@@ -95,15 +95,15 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
fDeviceCounter = fManager->Segment().find<DeviceCounter>(bipc::unique_instance).first;
if (fDeviceCounter)
{
LOG(DEBUG) << "shmem: device counter found, with value of " << fDeviceCounter->fCount << ". incrementing.";
LOG(debug) << "device counter found, with value of " << fDeviceCounter->fCount << ". incrementing.";
(fDeviceCounter->fCount)++;
LOG(DEBUG) << "shmem: incremented device counter, now: " << fDeviceCounter->fCount;
LOG(debug) << "incremented device counter, now: " << fDeviceCounter->fCount;
}
else
{
LOG(DEBUG) << "shmem: no device counter found, creating one and initializing with 1";
LOG(debug) << "no device counter found, creating one and initializing with 1";
fDeviceCounter = fManager->Segment().construct<DeviceCounter>(bipc::unique_instance)(1);
LOG(DEBUG) << "shmem: initialized device counter with: " << fDeviceCounter->fCount;
LOG(debug) << "initialized device counter with: " << fDeviceCounter->fCount;
}
// start shm monitor
@@ -112,24 +112,24 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
// MonitorStatus* monitorStatus = fManagementSegment.find<MonitorStatus>(bipc::unique_instance).first;
// if (monitorStatus == nullptr)
// {
// LOG(DEBUG) << "shmem: no shmmonitor found, starting...";
// LOG(debug) << "no shmmonitor found, starting...";
// StartMonitor();
// }
// else
// {
// LOG(DEBUG) << "shmem: found shmmonitor.";
// LOG(debug) << "found shmmonitor.";
// }
// }
// catch (std::exception& e)
// {
// LOG(ERROR) << "shmem: Exception during shmmonitor initialization: " << e.what() << ", application will now exit";
// LOG(error) << "Exception during shmmonitor initialization: " << e.what() << ", application will now exit";
// exit(EXIT_FAILURE);
// }
}
}
catch(bipc::interprocess_exception& e)
{
LOG(ERROR) << "Could not initialize shared memory transport: " << e.what();
LOG(error) << "Could not initialize shared memory transport: " << e.what();
throw runtime_error("Cannot update configuration. Socket method (bind/connect) not specified.");
}
@@ -143,20 +143,20 @@ void FairMQTransportFactorySHM::StartMonitor()
if (!bfs::exists(bfs::path("shmmonitor")))
{
LOG(ERROR) << "Could not find shmmonitor. Is it in the PATH? Monitor not started";
LOG(error) << "Could not find shmmonitor. Is it in the PATH? Monitor not started";
return;
}
// TODO: replace with Boost.Process once boost 1.64 is available
int r = system("shmmonitor --self-destruct &");
LOG(DEBUG) << r;
LOG(debug) << r;
do
{
MonitorStatus* monitorStatus = fManager->ManagementSegment().find<MonitorStatus>(bipc::unique_instance).first;
if (monitorStatus)
{
LOG(DEBUG) << "shmem: shmmonitor started";
LOG(debug) << "shmmonitor started";
break;
}
else
@@ -164,7 +164,7 @@ void FairMQTransportFactorySHM::StartMonitor()
this_thread::sleep_for(std::chrono::milliseconds(10));
if (++numTries > 100)
{
LOG(ERROR) << "Did not get response from shmmonitor after " << 10 * 100 << " milliseconds. Exiting.";
LOG(error) << "Did not get response from shmmonitor after " << 10 * 100 << " milliseconds. Exiting.";
exit(EXIT_FAILURE);
}
}
@@ -188,13 +188,13 @@ void FairMQTransportFactorySHM::SendHeartbeats()
}
else
{
LOG(DEBUG) << "control queue timeout";
LOG(debug) << "control queue timeout";
}
}
catch (bipc::interprocess_exception& ie)
{
this_thread::sleep_for(chrono::milliseconds(500));
// LOG(WARN) << "no " << controlQueueName << " found";
// LOG(warn) << "no " << controlQueueName << " found";
}
}
}
@@ -261,7 +261,7 @@ FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
{
if (errno == EINTR)
{
LOG(ERROR) << "shmem: failed closing context, reason: " << zmq_strerror(errno);
LOG(error) << "failed closing context, reason: " << zmq_strerror(errno);
}
else
{
@@ -272,7 +272,7 @@ FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
}
else
{
LOG(ERROR) << "shmem: Terminate(): context not available for shutdown";
LOG(error) << "context not available for shutdown";
}
bool lastRemoved = false;
@@ -284,14 +284,14 @@ FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
if (fDeviceCounter->fCount == 0)
{
LOG(DEBUG) << "shmem: last segment user, removing segment.";
LOG(debug) << "last segment user, removing segment.";
fManager->RemoveSegment();
lastRemoved = true;
}
else
{
LOG(DEBUG) << "shmem: other segment users present (" << fDeviceCounter->fCount << "), not removing it.";
LOG(debug) << "other segment users present (" << fDeviceCounter->fCount << "), not removing it.";
}
}

12
fairmq/shmem/FairMQUnmanagedRegionSHM.cxx
View File

@@ -25,15 +25,15 @@ FairMQUnmanagedRegionSHM::FairMQUnmanagedRegionSHM(Manager& manager, const size_
RegionCounter* rc = fManager.ManagementSegment().find<RegionCounter>(bipc::unique_instance).first;
if (rc)
{
LOG(DEBUG) << "shmem: region counter found, with value of " << rc->fCount << ". incrementing.";
LOG(debug) << "region counter found, with value of " << rc->fCount << ". incrementing.";
(rc->fCount)++;
LOG(DEBUG) << "shmem: incremented region counter, now: " << rc->fCount;
LOG(debug) << "incremented region counter, now: " << rc->fCount;
}
else
{
LOG(DEBUG) << "shmem: no region counter found, creating one and initializing with 1";
LOG(debug) << "no region counter found, creating one and initializing with 1";
rc = fManager.ManagementSegment().construct<RegionCounter>(bipc::unique_instance)(1);
LOG(DEBUG) << "shmem: initialized region counter with: " << rc->fCount;
LOG(debug) << "initialized region counter with: " << rc->fCount;
}
fRegionId = rc->fCount;
@@ -42,8 +42,8 @@ FairMQUnmanagedRegionSHM::FairMQUnmanagedRegionSHM(Manager& manager, const size_
}
catch (bipc::interprocess_exception& e)
{
LOG(ERROR) << "shmem: cannot create region. Already created/not cleaned up?";
LOG(ERROR) << e.what();
LOG(error) << "cannot create region. Already created/not cleaned up?";
LOG(error) << e.what();
exit(EXIT_FAILURE);
}
}

12
fairmq/shmem/Manager.cxx
View File

@@ -58,7 +58,7 @@ bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id,
auto it = fRegions.find(id);
if (it != fRegions.end())
{
LOG(ERROR) << "shmem: Trying to create a region that already exists";
LOG(error) << "Trying to create a region that already exists";
return nullptr;
}
else
@@ -88,7 +88,7 @@ Region* Manager::GetRemoteRegion(const uint64_t id)
}
catch (bipc::interprocess_exception& e)
{
// LOG(WARN) << "remote region (" << id << ") no longer exists";
// LOG(warn) << "remote region (" << id << ") no longer exists";
return nullptr;
}
@@ -104,20 +104,20 @@ void Manager::RemoveSegment()
{
if (bipc::shared_memory_object::remove(fSegmentName.c_str()))
{
LOG(DEBUG) << "shmem: successfully removed " << fSegmentName << " segment after the device has stopped.";
LOG(debug) << "successfully removed " << fSegmentName << " segment after the device has stopped.";
}
else
{
LOG(DEBUG) << "shmem: did not remove " << fSegmentName << " segment after the device stopped. Already removed?";
LOG(debug) << "did not remove " << fSegmentName << " segment after the device stopped. Already removed?";
}
if (bipc::shared_memory_object::remove(fManagementSegmentName.c_str()))
{
LOG(DEBUG) << "shmem: successfully removed '" << fManagementSegmentName << "' segment after the device has stopped.";
LOG(debug) << "successfully removed '" << fManagementSegmentName << "' segment after the device has stopped.";
}
else
{
LOG(DEBUG) << "shmem: did not remove '" << fManagementSegmentName << "' segment after the device stopped. Already removed?";
LOG(debug) << "did not remove '" << fManagementSegmentName << "' segment after the device stopped. Already removed?";
}
}

22
fairmq/shmem/Region.cxx
View File

@@ -38,19 +38,19 @@ Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, FairMQ
if (fRemote)
{
fShmemObject = bipc::shared_memory_object(bipc::open_only, fName.c_str(), bipc::read_write);
LOG(DEBUG) << "shmem: located remote region: " << fName;
LOG(debug) << "shmem: located remote region: " << fName;
fQueue = fair::mq::tools::make_unique<bipc::message_queue>(bipc::open_only, fQueueName.c_str());
LOG(DEBUG) << "shmem: located remote region queue: " << fQueueName;
LOG(debug) << "shmem: located remote region queue: " << fQueueName;
}
else
{
fShmemObject = bipc::shared_memory_object(bipc::create_only, fName.c_str(), bipc::read_write);
LOG(DEBUG) << "shmem: created region: " << fName;
LOG(debug) << "shmem: created region: " << fName;
fShmemObject.truncate(size);
fQueue = fair::mq::tools::make_unique<bipc::message_queue>(bipc::create_only, fQueueName.c_str(), 10000, sizeof(RegionBlock));
LOG(DEBUG) << "shmem: created region queue: " << fQueueName;
LOG(debug) << "shmem: created region queue: " << fQueueName;
}
fRegion = bipc::mapped_region(fShmemObject, bipc::read_write); // TODO: add HUGEPAGES flag here
// fRegion = bipc::mapped_region(fShmemObject, bipc::read_write, 0, 0, 0, MAP_HUGETLB | MAP_HUGE_1GB);
@@ -72,7 +72,7 @@ void Region::ReceiveAcks()
RegionBlock block;
if (fQueue->timed_receive(&block, sizeof(RegionBlock), recvdSize, priority, rcvTill))
{
// LOG(DEBUG) << "received: " << block.fHandle << " " << block.fSize << " " << block.fMessageId;
// LOG(debug) << "received: " << block.fHandle << " " << block.fSize << " " << block.fMessageId;
if (fCallback)
{
fCallback(reinterpret_cast<char*>(fRegion.get_address()) + block.fHandle, block.fSize, reinterpret_cast<void*>(block.fHint));
@@ -80,11 +80,11 @@ void Region::ReceiveAcks()
}
else
{
// LOG(DEBUG) << "queue " << fQueueName << " timeout!";
// LOG(debug) << "queue " << fQueueName << " timeout!";
}
} // while !fStop
LOG(DEBUG) << "worker for " << fName << " leaving.";
LOG(debug) << "worker for " << fName << " leaving.";
}
Region::~Region()
@@ -99,18 +99,18 @@ Region::~Region()
if (bipc::shared_memory_object::remove(fName.c_str()))
{
LOG(DEBUG) << "shmem: destroyed region " << fName;
LOG(debug) << "shmem: destroyed region " << fName;
}
if (bipc::message_queue::remove(fQueueName.c_str()))
{
LOG(DEBUG) << "shmem: removed region queue " << fName;
LOG(debug) << "shmem: removed region queue " << fName;
}
}
else
{
// LOG(DEBUG) << "shmem: region '" << fName << "' is remote, no cleanup necessary.";
LOG(DEBUG) << "shmem: region queue '" << fQueueName << "' is remote, no cleanup necessary";
// LOG(debug) << "shmem: region '" << fName << "' is remote, no cleanup necessary.";
LOG(debug) << "shmem: region queue '" << fQueueName << "' is remote, no cleanup necessary";
}
}

32
fairmq/shmem/prototype/FairMQShmPrototypeSampler.cxx
View File

@@ -40,11 +40,11 @@ FairMQShmPrototypeSampler::FairMQShmPrototypeSampler()
{
if (shared_memory_object::remove("FairMQSharedMemoryPrototype"))
{
LOG(INFO) << "Successfully removed shared memory upon device start.";
LOG(info) << "Successfully removed shared memory upon device start.";
}
else
{
LOG(INFO) << "Did not remove shared memory upon device start.";
LOG(info) << "Did not remove shared memory upon device start.";
}
}
@@ -52,11 +52,11 @@ FairMQShmPrototypeSampler::~FairMQShmPrototypeSampler()
{
if (shared_memory_object::remove("FairMQSharedMemoryPrototype"))
{
LOG(INFO) << "Successfully removed shared memory after the device has stopped.";
LOG(info) << "Successfully removed shared memory after the device has stopped.";
}
else
{
LOG(INFO) << "Did not remove shared memory after the device stopped. Still in use?";
LOG(info) << "Did not remove shared memory after the device stopped. Still in use?";
}
}
@@ -66,7 +66,7 @@ void FairMQShmPrototypeSampler::Init()
fMsgRate = fConfig->GetValue<int>("msg-rate");
SegmentManager::Instance().InitializeSegment("open_or_create", "FairMQSharedMemoryPrototype", 2000000000);
LOG(INFO) << "Created/Opened shared memory segment of 2,000,000,000 bytes. Available are "
LOG(info) << "Created/Opened shared memory segment of 2,000,000,000 bytes. Available are "
<< SegmentManager::Instance().Segment()->get_free_memory() << " bytes.";
}
@@ -75,7 +75,7 @@ void FairMQShmPrototypeSampler::Run()
// count sent messages (also used in creating ShmChunk container ID)
static uint64_t numSentMsgs = 0;
LOG(INFO) << "Starting the benchmark with message size of " << fMsgSize;
LOG(info) << "Starting the benchmark with message size of " << fMsgSize;
// start rate logger and acknowledgement listener in separate threads
thread rateLogger(&FairMQShmPrototypeSampler::Log, this, 1000);
@@ -123,7 +123,7 @@ void FairMQShmPrototypeSampler::Run()
// }
// catch (bipc::bad_alloc& ba)
// {
// LOG(WARN) << "Shared memory full...";
// LOG(warn) << "Shared memory full...";
// this_thread::sleep_for(chrono::milliseconds(100));
// continue;
// }
@@ -139,12 +139,12 @@ void FairMQShmPrototypeSampler::Run()
// charnum = 97;
// }
// LOG(DEBUG) << "chunk handle: " << owner->fPtr->GetHandle();
// LOG(DEBUG) << "chunk size: " << owner->fPtr->GetSize();
// LOG(DEBUG) << "owner (" << ownerID << ") use count: " << owner->fPtr.use_count();
// LOG(debug) << "chunk handle: " << owner->fPtr->GetHandle();
// LOG(debug) << "chunk size: " << owner->fPtr->GetSize();
// LOG(debug) << "owner (" << ownerID << ") use count: " << owner->fPtr.use_count();
// char* cptr = static_cast<char*>(ptr);
// LOG(DEBUG) << "check: " << cptr[3];
// LOG(debug) << "check: " << cptr[3];
// FairMQMessagePtr msg(NewSimpleMessage(ownerID));
@@ -156,9 +156,9 @@ void FairMQShmPrototypeSampler::Run()
metaPtr->fSize = fMsgSize;
metaPtr->fHandle = handle;
// LOG(INFO) << metaPtr->fSize;
// LOG(INFO) << metaPtr->fHandle;
// LOG(WARN) << ptr;
// LOG(info) << metaPtr->fSize;
// LOG(info) << metaPtr->fHandle;
// LOG(warn) << ptr;
if (Send(msg, "meta", 0) > 0)
{
@@ -180,7 +180,7 @@ void FairMQShmPrototypeSampler::Run()
// }
}
LOG(INFO) << "Sent " << numSentMsgs << " messages, leaving RUNNING state.";
LOG(info) << "Sent " << numSentMsgs << " messages, leaving RUNNING state.";
rateLogger.join();
// resetMsgCounter.join();
@@ -207,7 +207,7 @@ void FairMQShmPrototypeSampler::Log(const int intervalInMs)
msgPerSecOut = static_cast<double>(fMsgOutNew - fMsgOut) / static_cast<double>(msSinceLastLog) * 1000.;
fMsgOut = fMsgOutNew;
LOG(DEBUG) << fixed
LOG(debug) << fixed
<< setprecision(0) << "out: " << msgPerSecOut << " msg ("
<< setprecision(2) << mbPerSecOut << " MB)\t("
<< SegmentManager::Instance().Segment()->get_free_memory() / (1024. * 1024.) << " MB free)";

22
fairmq/shmem/prototype/FairMQShmPrototypeSink.cxx
View File

@@ -44,7 +44,7 @@ FairMQShmPrototypeSink::~FairMQShmPrototypeSink()
void FairMQShmPrototypeSink::Init()
{
SegmentManager::Instance().InitializeSegment("open_or_create", "FairMQSharedMemoryPrototype", 2000000000);
LOG(INFO) << "Created/Opened shared memory segment of 2,000,000,000 bytes. Available are "
LOG(info) << "Created/Opened shared memory segment of 2,000,000,000 bytes. Available are "
<< SegmentManager::Instance().Segment()->get_free_memory() << " bytes.";
}
@@ -65,9 +65,9 @@ void FairMQShmPrototypeSink::Run()
bipc::managed_shared_memory::handle_t handle = hdr->fHandle;
void* ptr = SegmentManager::Instance().Segment()->get_address_from_handle(handle);
// LOG(INFO) << size;
// LOG(INFO) << handle;
// LOG(WARN) << ptr;
// LOG(info) << size;
// LOG(info) << handle;
// LOG(warn) << ptr;
fBytesInNew += size;
++fMsgInNew;
@@ -78,21 +78,21 @@ void FairMQShmPrototypeSink::Run()
// find the shared pointer in shared memory with its ID
// ShPtrOwner* owner = SegmentManager::Instance().Segment()->find<ShPtrOwner>(ownerID.c_str()).first;
// LOG(DEBUG) << "owner (" << ownerID << ") use count: " << owner->fPtr.use_count();
// LOG(debug) << "owner (" << ownerID << ") use count: " << owner->fPtr.use_count();
// if (owner)
// {
// // void* ptr = owner->fPtr->GetData();
// // LOG(DEBUG) << "chunk handle: " << owner->fPtr->GetHandle();
// // LOG(DEBUG) << "chunk size: " << owner->fPtr->GetSize();
// // LOG(debug) << "chunk handle: " << owner->fPtr->GetHandle();
// // LOG(debug) << "chunk size: " << owner->fPtr->GetSize();
// fBytesInNew += owner->fPtr->GetSize();
// ++fMsgInNew;
// // char* cptr = static_cast<char*>(ptr);
// // LOG(DEBUG) << "check: " << cptr[3];
// // LOG(debug) << "check: " << cptr[3];
// SegmentManager::Instance().Segment()->deallocate(ptr);
@@ -100,7 +100,7 @@ void FairMQShmPrototypeSink::Run()
// }
// else
// {
// LOG(WARN) << "Shared pointer is zero.";
// LOG(warn) << "Shared pointer is zero.";
// }
@@ -108,7 +108,7 @@ void FairMQShmPrototypeSink::Run()
}
}
LOG(INFO) << "Received " << numReceivedMsgs << " messages, leaving RUNNING state.";
LOG(info) << "Received " << numReceivedMsgs << " messages, leaving RUNNING state.";
rateLogger.join();
}
@@ -134,7 +134,7 @@ void FairMQShmPrototypeSink::Log(const int intervalInMs)
msgPerSecIn = static_cast<double>(fMsgInNew - fMsgIn) / static_cast<double>(msSinceLastLog) * 1000.;
fMsgIn = fMsgInNew;
LOG(DEBUG) << fixed
LOG(debug) << fixed
<< setprecision(0) << "in: " << msgPerSecIn << " msg ("
<< setprecision(2) << mbPerSecIn << " MB)\t("
<< SegmentManager::Instance().Segment()->get_free_memory() / (1024. * 1024.) << " MB free)";

14
fairmq/shmem/prototype/ShmChunk.h
View File

@@ -64,13 +64,13 @@ class SegmentManager
{
if (++numTries == 5)
{
LOG(ERROR) << "Could not open shared memory after " << numTries << " attempts, exiting!";
LOG(error) << "Could not open shared memory after " << numTries << " attempts, exiting!";
exit(EXIT_FAILURE);
}
else
{
LOG(DEBUG) << "Could not open shared memory segment on try " << numTries << ". Retrying in 1 second...";
LOG(DEBUG) << ie.what();
LOG(debug) << "Could not open shared memory segment on try " << numTries << ". Retrying in 1 second...";
LOG(debug) << ie.what();
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
@@ -80,18 +80,18 @@ class SegmentManager
}
else
{
LOG(ERROR) << "Unknown operation when initializing shared memory segment: " << op;
LOG(error) << "Unknown operation when initializing shared memory segment: " << op;
}
}
catch (std::exception& e)
{
LOG(ERROR) << "Exception during shared memory segment initialization: " << e.what() << ", application will now exit";
LOG(error) << "Exception during shared memory segment initialization: " << e.what() << ", application will now exit";
exit(EXIT_FAILURE);
}
}
else
{
LOG(INFO) << "Segment already initialized";
LOG(info) << "Segment already initialized";
}
}
@@ -103,7 +103,7 @@ class SegmentManager
}
else
{
LOG(ERROR) << "Segment not initialized";
LOG(error) << "Segment not initialized";
exit(EXIT_FAILURE);
}
}