|
9 | 9 | // these subcomponents is subject to the terms and conditions of the subcomponent's license, as noted in the LICENSE |
10 | 10 | // file. |
11 | 11 |
|
12 | | -#include <cstring> |
13 | | - |
14 | 12 | #include "MessageBase.hpp" |
15 | | - |
16 | 13 | #include "log/logger.hpp" |
17 | 14 | #include "util/assertUtils.hpp" |
18 | 15 | #include "ReplicaConfig.hpp" |
19 | 16 |
|
| 17 | +#include <cstring> |
| 18 | +#include <sstream> |
| 19 | + |
20 | 20 | #ifdef DEBUG_MEMORY_MSG |
21 | 21 | #include <set> |
22 | 22 |
|
@@ -46,11 +46,66 @@ void MessageBase::printLiveMessages() { |
46 | 46 | namespace bftEngine { |
47 | 47 | namespace impl { |
48 | 48 |
|
| 49 | +// static class members for diagnostics server |
| 50 | +std::array<std::atomic<size_t>, MsgCode::LastMsgCodeVal> MessageBase::Statistics::AliveIncomingExtrnMsgsBufs{}; |
| 51 | +static_assert(MsgCode::LastMsgCodeVal < 2000, |
| 52 | + "MessageBase AliveIncomingExtrnMsgsBufs array too big (above 2000), check MsgCode enum definition"); |
| 53 | +Bitmap MessageBase::Statistics::IncomingExtrnMsgReceivedAtLeastOnceFlags{MsgCode::LastMsgCodeVal}; |
| 54 | +std::mutex MessageBase::Statistics::messagesStatsMonitoringMutex_{}; |
| 55 | +std::atomic<size_t> MessageBase::Statistics::numIncomingExtrnMsgsBufAllocs{0}; |
| 56 | +std::atomic<size_t> MessageBase::Statistics::numIncomingExtrnMsgsBufFrees{0}; |
| 57 | +// End of static class members for diagnostics server |
| 58 | + |
49 | 59 | MessageBase::~MessageBase() { |
50 | 60 | #ifdef DEBUG_MEMORY_MSG |
51 | 61 | liveMessagesDebug.erase(this); |
52 | 62 | #endif |
53 | | - if (owner_) std::free((char *)msgBody_); |
| 63 | + |
| 64 | + // if this obj is not the owner of the buf (owner_ is false), the buf may have already be freed - hence access to it |
| 65 | + // is prohibited |
| 66 | + if (owner_) { |
| 67 | + if (isIncomingMsg_) { |
| 68 | + MsgCode::Type msg_code = static_cast<MsgCode::Type>(msgBody_->msgType); |
| 69 | + MessageBase::Statistics::updateDiagnosticsCountersOnBufRelease(msg_code); |
| 70 | + } |
| 71 | + std::free((char *)msgBody_); |
| 72 | + } |
| 73 | +} |
| 74 | + |
| 75 | +void MessageBase::Statistics::updateDiagnosticsCountersOnBufAlloc(MsgCode::Type msg_code) { |
| 76 | + if (!MessageBase::Statistics::IncomingExtrnMsgReceivedAtLeastOnceFlags.get(msg_code)) { |
| 77 | + // here is a very rare event - first time ever receiving msg of type msg_code, hence |
| 78 | + // using a lock here shouldn't affect performnce. in other places we avoid locks where possible. |
| 79 | + // It is possible that two threads will call the "set" function but it's ok since it's |
| 80 | + // just a flag telling us if the message was received at least once. setting the bit twice causes |
| 81 | + // no damage. |
| 82 | + std::lock_guard<std::mutex> lock(MessageBase::Statistics::messagesStatsMonitoringMutex_); |
| 83 | + MessageBase::Statistics::IncomingExtrnMsgReceivedAtLeastOnceFlags.set(msg_code); |
| 84 | + } |
| 85 | + |
| 86 | + // use "++" operator to ensure atomicity |
| 87 | + MessageBase::Statistics::numIncomingExtrnMsgsBufAllocs++; |
| 88 | + MessageBase::Statistics::AliveIncomingExtrnMsgsBufs[msg_code]++; |
| 89 | +} |
| 90 | + |
| 91 | +void MessageBase::Statistics::updateDiagnosticsCountersOnBufRelease(MsgCode::Type msg_code) { |
| 92 | + if (MessageBase::Statistics::AliveIncomingExtrnMsgsBufs[msg_code] == 0) { |
| 93 | + LOG_ERROR(GL, "Trying to dec a counter of a msg that hasn't been inserted yet, msg code: " << (msg_code)); |
| 94 | + return; |
| 95 | + } else { |
| 96 | + // use "--" operator to ensure atomicity |
| 97 | + MessageBase::Statistics::AliveIncomingExtrnMsgsBufs[msg_code]--; |
| 98 | + } |
| 99 | + // use "++" operator to ensure atomicity |
| 100 | + MessageBase::Statistics::numIncomingExtrnMsgsBufFrees++; |
| 101 | +} |
| 102 | + |
| 103 | +void MessageBase::releaseOwnership() { |
| 104 | + if (!owner_) { |
| 105 | + LOG_ERROR(GL, "Trying to release ownership of a MessageBase obj that is already not the buffer owner"); |
| 106 | + } else { |
| 107 | + owner_ = false; |
| 108 | + } |
54 | 109 | } |
55 | 110 |
|
56 | 111 | void MessageBase::shrinkToFit() { |
@@ -102,13 +157,17 @@ MessageBase::MessageBase(NodeIdType sender, MsgType type, SpanContextSize spanCo |
102 | 157 | #endif |
103 | 158 | } |
104 | 159 |
|
105 | | -MessageBase::MessageBase(NodeIdType sender, MessageBase::Header *body, MsgSize size, bool ownerOfStorage) { |
106 | | - msgBody_ = body; |
107 | | - msgSize_ = size; |
108 | | - storageSize_ = size; |
109 | | - sender_ = sender; |
110 | | - owner_ = ownerOfStorage; |
111 | | - |
| 160 | +MessageBase::MessageBase(NodeIdType sender, MessageBase::Header *body, MsgSize size, bool ownerOfStorage) |
| 161 | + : MessageBase(sender, body, size, ownerOfStorage, false) {} |
| 162 | + |
| 163 | +MessageBase::MessageBase( |
| 164 | + NodeIdType sender, MessageBase::Header *body, MsgSize size, bool ownerOfStorage, bool isIncoming) |
| 165 | + : msgBody_(body), |
| 166 | + msgSize_(size), |
| 167 | + storageSize_(size), |
| 168 | + sender_(sender), |
| 169 | + owner_(ownerOfStorage), |
| 170 | + isIncomingMsg_(isIncoming) { |
112 | 171 | #ifdef DEBUG_MEMORY_MSG |
113 | 172 | liveMessagesDebug.insert(this); |
114 | 173 | #endif |
@@ -235,5 +294,38 @@ MessageBase *MessageBase::deserializeMsg(char *&buf, size_t bufLen, size_t &actu |
235 | 294 | return msg; |
236 | 295 | } |
237 | 296 |
|
| 297 | +// Start methods for diagnostics server: |
| 298 | +std::string MessageBase::Statistics::getNumBuffsAllocatedForExtrnIncomingMsgs() { |
| 299 | + return (std::string(" Num buffer allocations for external incoming messages: " + |
| 300 | + std::to_string(MessageBase::Statistics::numIncomingExtrnMsgsBufAllocs))); |
| 301 | +} |
| 302 | +std::string MessageBase::Statistics::getNumBuffsFreedForExtrnIncomingMsgs() { |
| 303 | + return (std::string(" Num buffer frees for external incoming messages: " + |
| 304 | + std::to_string(MessageBase::Statistics::numIncomingExtrnMsgsBufFrees))); |
| 305 | +} |
| 306 | +std::string MessageBase::Statistics::getNumAliveExtrnIncomingMsgsObjsPerType() { |
| 307 | + std::ostringstream oss; |
| 308 | + oss << " Alive extrnal incoming message objects per type: " << std::endl; |
| 309 | + oss << " --------------------------------------- " << std::endl; |
| 310 | + |
| 311 | + for (int i = 0; i < MsgCode::LastMsgCodeVal; ++i) { |
| 312 | + bool wasMsgEverRecieved; |
| 313 | + |
| 314 | + { |
| 315 | + std::lock_guard<std::mutex> lock(messagesStatsMonitoringMutex_); |
| 316 | + wasMsgEverRecieved = MessageBase::Statistics::IncomingExtrnMsgReceivedAtLeastOnceFlags.get(i); |
| 317 | + } |
| 318 | + |
| 319 | + if (wasMsgEverRecieved) { |
| 320 | + oss << " " << static_cast<MsgCode::Type>(i) << ": " << MessageBase::Statistics::AliveIncomingExtrnMsgsBufs[i] |
| 321 | + << std::endl; |
| 322 | + } |
| 323 | + } |
| 324 | + |
| 325 | + oss << " ** All messages not on the list were never received by replica ** "; |
| 326 | + return oss.str(); |
| 327 | +} |
| 328 | +// End methods for diagnostics server ^ |
| 329 | + |
238 | 330 | } // namespace impl |
239 | 331 | } // namespace bftEngine |
0 commit comments