diff --git a/indra/llmessage/CMakeLists.txt b/indra/llmessage/CMakeLists.txt index 62897515dd..46b9bec6a8 100644 --- a/indra/llmessage/CMakeLists.txt +++ b/indra/llmessage/CMakeLists.txt @@ -228,6 +228,7 @@ if (LL_TESTS) LL_ADD_INTEGRATION_TEST(llsdmessagebuilder "" "${test_libs}") LL_ADD_INTEGRATION_TEST(llsdmessagereader "" "${test_libs}") LL_ADD_INTEGRATION_TEST(llmessageconfig "" "${test_libs}") + LL_ADD_INTEGRATION_TEST(llmessagetemplate "" "${test_libs}") LL_ADD_INTEGRATION_TEST(llmessagetemplateparser "" "${test_libs}") LL_ADD_INTEGRATION_TEST(llpartdata "" "${test_libs}") LL_ADD_INTEGRATION_TEST(llservicebuilder "" "${test_libs}") diff --git a/indra/llmessage/llcircuit.cpp b/indra/llmessage/llcircuit.cpp index ae7c555d57..056441ba21 100644 --- a/indra/llmessage/llcircuit.cpp +++ b/indra/llmessage/llcircuit.cpp @@ -421,6 +421,8 @@ S32 LLCircuitData::resendUnackedPackets(const F64Seconds now) LLCircuit::LLCircuit(const F32Seconds circuit_heartbeat_interval, const F32Seconds circuit_timeout) : mLastCircuit(NULL), + mCircuitsUnusedCount(0), + mDeferredDeletion(false), mHeartbeatInterval(circuit_heartbeat_interval), mHeartbeatTimeout(circuit_timeout) {} @@ -433,6 +435,13 @@ LLCircuit::~LLCircuit() llcompose1( DeletePointerFunctor(), llselect2nd())); + + // The receiver thread is long gone by now, nothing can be using these. + for (auto& dead : mGraveyard) + { + delete dead.first; + } + mGraveyard.clear(); } LLCircuitData *LLCircuit::addCircuitData(const LLHost &host, TPACKETID in_id) @@ -452,36 +461,89 @@ LLCircuitData *LLCircuit::addCircuitData(const LLHost &host, TPACKETID in_id) void LLCircuit::removeCircuitData(const LLHost &host) { LL_INFOS() << "LLCircuit::removeCircuitData for " << host << LL_ENDL; - std::lock_guard lock(mCircuitMutex); - mLastCircuit = NULL; - circuit_data_map::iterator it = mCircuitData.find(host); - if(it != mCircuitData.end()) + LLCircuitData *cdp = NULL; { - LLCircuitData *cdp = it->second; - mCircuitData.erase(it); + std::lock_guard lock(mCircuitMutex); + circuit_data_map::iterator it = mCircuitData.find(host); + if(it != mCircuitData.end()) + { + cdp = it->second; + mCircuitData.erase(it); + + LLCircuit::ping_set_t::iterator psit = mPingSet.find(cdp); + if (psit != mPingSet.end()) + { + mPingSet.erase(psit); + } + else + { + LL_WARNS() << "Couldn't find entry for next ping in ping set!" << LL_ENDL; + } + + // Clean up from optimization maps + mUnackedCircuitMap.erase(host); + mSendAckMap.erase(host); + + if (mDeferredDeletion) + { + // The UDP receiver thread may be decoding a packet on this + // circuit right now, and it uses plain pointers. The circuit + // is no longer reachable from any of the maps, so nobody can + // pick it up anymore: bury it and let cleanupGraveyard() + // delete it once the receiver thread is done with it. + mGraveyard.emplace_back(cdp, mCircuitsUnusedCount.load()); + cdp = NULL; + } + } + + // This also has to happen AFTER we nuke the circuit, because various + // callbacks for the circuit may result in messages being sent to + // this circuit, and the setting of mLastCircuit. We don't check + // if the host matches, but we don't really care because mLastCircuit + // is an optimization, and this happens VERY rarely. + mLastCircuit = NULL; + } + + // Deleting a circuit fires reliable packet callbacks, so it has to happen + // with mCircuitMutex released. + delete cdp; + + cleanupGraveyard(); +} - LLCircuit::ping_set_t::iterator psit = mPingSet.find(cdp); - if (psit != mPingSet.end()) +void LLCircuit::cleanupGraveyard() +{ + // Deleted with mCircuitMutex released, see removeCircuitData(). + std::vector doomed; + { + std::lock_guard lock(mCircuitMutex); + if (mGraveyard.empty()) { - mPingSet.erase(psit); + return; } - else + // The receiver thread bumps mCircuitsUnusedCount between packets, so + // a different value means it has been between packets since the entry + // was buried and can't be holding a pointer to it. + bool thread_gone = !mDeferredDeletion; + U32 unused_count = mCircuitsUnusedCount.load(); + for (auto it = mGraveyard.begin(); it != mGraveyard.end(); ) { - LL_WARNS() << "Couldn't find entry for next ping in ping set!" << LL_ENDL; + if (thread_gone || unused_count != it->second) + { + doomed.push_back(it->first); + it = mGraveyard.erase(it); + } + else + { + ++it; + } } - - // Clean up from optimization maps - mUnackedCircuitMap.erase(host); - mSendAckMap.erase(host); - delete cdp; } - // This also has to happen AFTER we nuke the circuit, because various - // callbacks for the circuit may result in messages being sent to - // this circuit, and the setting of mLastCircuit. We don't check - // if the host matches, but we don't really care because mLastCircuit - // is an optimization, and this happens VERY rarely. - mLastCircuit = NULL; + for (LLCircuitData* dead : doomed) + { + delete dead; + } } void LLCircuitData::setAlive(bool b_alive) @@ -871,6 +933,9 @@ void LLCircuit::updateWatchDogTimers(LLMessageSystem *msgsys) } } } + + // Delete circuits that were buried while the receiver thread was busy. + cleanupGraveyard(); } @@ -1107,16 +1172,16 @@ bool LLCircuitData::checkCircuitTimeout() // correctly place the packet in the correct list to be acked later. bool LLCircuitData::collectRAck(TPACKETID packet_num) { + // Global lock order: LLCircuit::mCircuitMutex before mDataMutex. + std::lock_guard circuit_lock(gMessageSystem->mCircuitInfo.mCircuitMutex); std::lock_guard data_lock(mDataMutex); if (mAcks.empty()) { - std::lock_guard circuit_lock(gMessageSystem->mCircuitInfo.mCircuitMutex); // First extra ack, we need to add ourselves to the list of circuits that need to send acks gMessageSystem->mCircuitInfo.mSendAckMap[mHost] = this; } mAcks.push_back(packet_num); - std::lock_guard circuit_lock(gMessageSystem->mCircuitInfo.mCircuitMutex); if (mAckCreationTime == 0) { mAckCreationTime = getAgeInSeconds(); @@ -1129,7 +1194,8 @@ bool LLCircuitData::collectRAck(TPACKETID packet_num) void LLCircuit::sendAcks(F32 collect_time) { collect_time = llclamp(collect_time, 0.f, LL_COLLECT_ACK_TIME_MAX); - std::vector circuits_to_flush; // Collect while holding mCircuitMutex + // Collect while holding mCircuitMutex, send with no lock held. + std::vector > > circuits_to_flush; { std::lock_guard circuit_lock(mCircuitMutex); circuit_data_map::iterator it = mSendAckMap.begin(); @@ -1137,30 +1203,35 @@ void LLCircuit::sendAcks(F32 collect_time) { circuit_data_map::iterator cur_it = it++; LLCircuitData* cd = (*cur_it).second; - std::lock_guard data_lock(cd->mDataMutex); - S32 count = (S32)cd->mAcks.size(); - F32 age = cd->getAgeInSeconds() - cd->mAckCreationTime; - if (age > collect_time || count == 0) + std::vector acks; { - if (count > 0) + std::lock_guard data_lock(cd->mDataMutex); + S32 count = (S32)cd->mAcks.size(); + F32 age = cd->getAgeInSeconds() - cd->mAckCreationTime; + if (age <= collect_time && count != 0) { - circuits_to_flush.push_back(cd); + continue; } + // Take the acks while both locks are held, so that a circuit + // removed from mSendAckMap never keeps pending acks behind: + // any ack collected after this point sees an empty mAcks and + // re-registers the circuit. + acks.swap(cd->mAcks); + cd->mAckCreationTime = 0.f; // remove data map mSendAckMap.erase(cur_it); } + if (!acks.empty()) + { + circuits_to_flush.emplace_back(cd, std::move(acks)); + } } } - for (LLCircuitData* cd : circuits_to_flush) + for (const auto& flush : circuits_to_flush) { - std::vector acks; - { - std::lock_guard data_lock(cd->mDataMutex); - acks.swap(cd->mAcks); - cd->mAckCreationTime = 0.f; - } - + LLCircuitData* cd = flush.first; + const std::vector& acks = flush.second; S32 count = (S32)acks.size(); // send the packet acks diff --git a/indra/llmessage/llcircuit.h b/indra/llmessage/llcircuit.h index 959ff838c6..cb06e9c53a 100644 --- a/indra/llmessage/llcircuit.h +++ b/indra/llmessage/llcircuit.h @@ -28,7 +28,9 @@ #ifndef LL_LLCIRCUIT_H #define LL_LLCIRCUIT_H +#include #include +#include #include #include "llerror.h" @@ -73,7 +75,16 @@ class LLSD; // Classes // - +// LOCK ORDER +// +// Two mutexes guard circuit state: LLCircuit::mCircuitMutex (the circuit maps) +// and LLCircuitData::mDataMutex (per-circuit data). Whenever both are needed, +// they must be acquired in this order: +// +// LLCircuit::mCircuitMutex -> LLCircuitData::mDataMutex +// +// Never acquire mCircuitMutex while holding a circuit's mDataMutex, and never +// run callbacks or send messages while holding either of them. class LLCircuitData { public: @@ -188,6 +199,8 @@ class LLCircuitData // Call this method when a reliable message comes in - this will // correctly place the packet in the correct list to be acked // later. RAack = requested ack + // Takes LLCircuit::mCircuitMutex and then mDataMutex, so callers must not + // hold either of them (see the lock order note above). bool collectRAck(TPACKETID packet_num); @@ -305,13 +318,30 @@ class LLCircuit ~LLCircuit(); // ACCESSORS + // The returned circuit stays valid for as long as the caller keeps the + // main thread from running removeCircuitData(); on the UDP receiver + // thread it stays valid until the next markCircuitsUnused(). LLCircuitData* findCircuit(const LLHost& host) const; bool isCircuitAlive(const LLHost& host) const; // MANIPULATORS LLCircuitData *addCircuitData(const LLHost &host, TPACKETID in_id); + // Main thread only. While the UDP receiver thread is running, the circuit + // is parked in the graveyard instead of being deleted right away, since + // that thread may still be decoding a packet on it. void removeCircuitData(const LLHost &host); + // Called from the main thread, deletes the circuits in the graveyard that + // the UDP receiver thread is guaranteed to be done with. + void cleanupGraveyard(); + // Called from the UDP receiver thread at a point where it does not hold + // any LLCircuitData pointer, so that circuits removed before this point + // can be deleted by the main thread. + void markCircuitsUnused() { mCircuitsUnusedCount.fetch_add(1); } + // Enables the graveyard. Called from the main thread when the UDP + // receiver thread is started and after it has been joined. + void setDeferredDeletion(bool defer) { mDeferredDeletion = defer; } + void updateWatchDogTimers(LLMessageSystem *msgsys); void resendUnackedPackets(S32& unacked_list_length, S32& unacked_list_size); @@ -346,6 +376,16 @@ class LLCircuit // set in otherwise const methods, so it is declared mutable. mutable LLCircuitData* mLastCircuit; + // Circuits that were removed from the maps while the UDP receiver thread + // might still have been using them, with the value mCircuitsUnusedCount + // had at the time of removal. Guarded by mCircuitMutex. + std::vector > mGraveyard; + // Bumped by the receiver thread whenever it is between packets and hence + // holds no circuit pointer. A graveyard entry may be deleted as soon as + // this has moved past the value stored with it. + std::atomic mCircuitsUnusedCount; + std::atomic mDeferredDeletion; + private: const F32Seconds mHeartbeatInterval; const F32Seconds mHeartbeatTimeout; diff --git a/indra/llmessage/llmessagetemplate.h b/indra/llmessage/llmessagetemplate.h index 2cf1da7119..188b94bb80 100644 --- a/indra/llmessage/llmessagetemplate.h +++ b/indra/llmessage/llmessagetemplate.h @@ -31,6 +31,7 @@ #include "llstl.h" #include "llindexedvector.h" #include "llmsgvariabletype.h" +#include #include class LLMsgVarData @@ -292,21 +293,81 @@ class LLMessageTemplate mDeprecation(MD_NOTDEPRECATED), mMessageNumber(message_number), mTotalSize(0), + mBanFromTrusted(false), + mBanFromUntrusted(false), mReceiveCount(0), mReceiveBytes(0), mReceiveInvalid(0), mTotalDecoded(0), mTotalDecodeTime(0.f), mMaxDecodeTimePerMsg(0.f), - mBanFromTrusted(false), - mBanFromUntrusted(false), mHandlerFunc(NULL), - mHandleOnUdpThread(false), - mUserData(NULL) + mUserData(NULL), + mHandleOnUdpThread(false) { mName = LLMessageStringTable::getInstance()->getString(name); } + // The counters below are atomic, which makes LLMessageTemplate neither + // implicitly copyable nor movable. Spell the copy out instead, to keep + // the (shallow) copy semantics this class has always had. + LLMessageTemplate(const LLMessageTemplate& rhs) + : + mMemberBlocks(rhs.mMemberBlocks), + mName(rhs.mName), + mFrequency(rhs.mFrequency), + mTrust(rhs.mTrust), + mEncoding(rhs.mEncoding), + mDeprecation(rhs.mDeprecation), + mMessageNumber(rhs.mMessageNumber), + mTotalSize(rhs.mTotalSize), + mBanFromTrusted(rhs.mBanFromTrusted), + mBanFromUntrusted(rhs.mBanFromUntrusted), + mReceiveCount(rhs.getReceiveCount()), + mReceiveBytes(rhs.getReceiveBytes()), + mReceiveInvalid(rhs.getReceiveInvalid()), + mTotalDecoded(rhs.getTotalDecoded()), + mTotalDecodeTime(rhs.getTotalDecodeTime()), + mMaxDecodeTimePerMsg(rhs.getMaxDecodeTimePerMsg()), + mHandlerFunc(NULL), + mUserData(NULL), + mHandleOnUdpThread(false) + { + std::shared_lock lock(sHandlerMutex); + mHandlerFunc = rhs.mHandlerFunc; + mUserData = rhs.mUserData; + mHandleOnUdpThread = rhs.mHandleOnUdpThread; + } + + LLMessageTemplate& operator=(const LLMessageTemplate& rhs) + { + if (this != &rhs) + { + mMemberBlocks = rhs.mMemberBlocks; + mName = rhs.mName; + mFrequency = rhs.mFrequency; + mTrust = rhs.mTrust; + mEncoding = rhs.mEncoding; + mDeprecation = rhs.mDeprecation; + mMessageNumber = rhs.mMessageNumber; + mTotalSize = rhs.mTotalSize; + mBanFromTrusted = rhs.mBanFromTrusted; + mBanFromUntrusted = rhs.mBanFromUntrusted; + mReceiveCount.store(rhs.getReceiveCount(), std::memory_order_relaxed); + mReceiveBytes.store(rhs.getReceiveBytes(), std::memory_order_relaxed); + mReceiveInvalid.store(rhs.getReceiveInvalid(), std::memory_order_relaxed); + mTotalDecoded.store(rhs.getTotalDecoded(), std::memory_order_relaxed); + mTotalDecodeTime.store(rhs.getTotalDecodeTime(), std::memory_order_relaxed); + mMaxDecodeTimePerMsg.store(rhs.getMaxDecodeTimePerMsg(), std::memory_order_relaxed); + + std::unique_lock lock(sHandlerMutex); + mHandlerFunc = rhs.mHandlerFunc; + mUserData = rhs.mUserData; + mHandleOnUdpThread = rhs.mHandleOnUdpThread; + } + return *this; + } + ~LLMessageTemplate() { for_each(mMemberBlocks.begin(), mMemberBlocks.end(), DeletePointer()); @@ -456,6 +517,48 @@ class LLMessageTemplate return mDeprecation == MD_UDPBLACKLISTED; } + // Receive counters. Incremented by message validation, which happens on + // LLUDPReceiverThread, and read/reset by main-thread reporting, so all + // accesses go through atomics. + void recordReceive() { mReceiveCount.fetch_add(1, std::memory_order_relaxed); } + void recordReceiveCount(U32 bytes, bool invalid) + { + mReceiveCount.fetch_add(1, std::memory_order_relaxed); + mReceiveBytes.fetch_add(bytes, std::memory_order_relaxed); + if (invalid) + { + mReceiveInvalid.fetch_add(1, std::memory_order_relaxed); + } + } + void resetReceiveCounts() + { + mReceiveCount.store(0, std::memory_order_relaxed); + mReceiveBytes.store(0, std::memory_order_relaxed); + mReceiveInvalid.store(0, std::memory_order_relaxed); + } + U32 getReceiveCount() const { return mReceiveCount.load(std::memory_order_relaxed); } + U32 getReceiveBytes() const { return mReceiveBytes.load(std::memory_order_relaxed); } + U32 getReceiveInvalid() const { return mReceiveInvalid.load(std::memory_order_relaxed); } + + // Decode timing statistics. Messages are dispatched both from the main + // thread and from LLUDPReceiverThread, and the statistics are read and + // reset by main-thread reporting, so all accesses go through atomics. + void recordDecodeTime(F32 decode_time) + { + mTotalDecoded.fetch_add(1, std::memory_order_relaxed); + atomicAddF32(mTotalDecodeTime, decode_time); + atomicMaxF32(mMaxDecodeTimePerMsg, decode_time); + } + void resetDecodeStats() + { + mTotalDecoded.store(0, std::memory_order_relaxed); + mTotalDecodeTime.store(0.f, std::memory_order_relaxed); + mMaxDecodeTimePerMsg.store(0.f, std::memory_order_relaxed); + } + U32 getTotalDecoded() const { return mTotalDecoded.load(std::memory_order_relaxed); } + F32 getTotalDecodeTime() const { return mTotalDecodeTime.load(std::memory_order_relaxed); } + F32 getMaxDecodeTimePerMsg() const { return mMaxDecodeTimePerMsg.load(std::memory_order_relaxed); } + void banUdp(); bool isBanned(bool trustedSource) const @@ -481,17 +584,40 @@ class LLMessageTemplate EMsgDeprecation mDeprecation; U32 mMessageNumber; S32 mTotalSize; - U32 mReceiveCount; // how many of this template have been received since last reset - U32 mReceiveBytes; // How many bytes received - U32 mReceiveInvalid; // How many "invalid" packets - U32 mTotalDecoded; // Total messages successfully decoded - F32 mTotalDecodeTime; // Total time successfully decoding messages - F32 mMaxDecodeTimePerMsg; - bool mBanFromTrusted; bool mBanFromUntrusted; private: + // Written from LLUDPReceiverThread, read and reset from the main thread. + // Use the record*()/get*()/reset*() accessors above. + std::atomic mReceiveCount; // how many of this template have been received since last reset + std::atomic mReceiveBytes; // How many bytes received + std::atomic mReceiveInvalid; // How many "invalid" packets + std::atomic mTotalDecoded; // Total messages successfully decoded + std::atomic mTotalDecodeTime; // Total time successfully decoding messages + std::atomic mMaxDecodeTimePerMsg; + + // std::atomic has no portable fetch_add()/fetch_max(), so do it + // with the usual compare-exchange loop. + static void atomicAddF32(std::atomic& value, F32 addend) + { + F32 current = value.load(std::memory_order_relaxed); + while (!value.compare_exchange_weak(current, current + addend, std::memory_order_relaxed)) + { + // 'current' has been updated with the current value, retry. + } + } + + static void atomicMaxF32(std::atomic& value, F32 candidate) + { + F32 current = value.load(std::memory_order_relaxed); + while (current < candidate + && !value.compare_exchange_weak(current, candidate, std::memory_order_relaxed)) + { + // 'current' has been updated with the current value, retry. + } + } + // Since setHandlerFunc are mostly startup specific, one-time init, // a shared mutex is enough for all templates static inline std::shared_mutex sHandlerMutex; diff --git a/indra/llmessage/lltemplatemessagereader.cpp b/indra/llmessage/lltemplatemessagereader.cpp index cdb2b59745..d1b3855f2e 100644 --- a/indra/llmessage/lltemplatemessagereader.cpp +++ b/indra/llmessage/lltemplatemessagereader.cpp @@ -755,7 +755,7 @@ bool LLTemplateMessageReader::validateMessage(const U8* buffer, bool valid = decodeTemplate(buffer, buffer_size, &mCurrentRMessageTemplate ); if(valid) { - mCurrentRMessageTemplate->mReceiveCount++; + mCurrentRMessageTemplate->recordReceive(); //LL_DEBUGS() << "MessageRecvd:" // << mCurrentRMessageTemplate->mName // << " from " << sender << LL_ENDL; diff --git a/indra/llmessage/lludpreceiverthread.cpp b/indra/llmessage/lludpreceiverthread.cpp index b9222d492d..8eb12a1c96 100644 --- a/indra/llmessage/lludpreceiverthread.cpp +++ b/indra/llmessage/lludpreceiverthread.cpp @@ -61,6 +61,11 @@ void LLUDPReceiverThread::run() { LL_PROFILE_ZONE_SCOPED_CATEGORY_NETWORK; + // Nothing is being decoded at this point, so no circuit pointer is + // held: tell the main thread that circuits removed before now can be + // deleted (see LLCircuit::cleanupGraveyard()). + gMessageSystem->mCircuitInfo.markCircuitsUnused(); + LLHost invalid_host; LLPacketBuffer pkt(invalid_host, nullptr, 0); S32 packet_size = 0; diff --git a/indra/llmessage/message.cpp b/indra/llmessage/message.cpp index d96bc34727..924a6a3494 100644 --- a/indra/llmessage/message.cpp +++ b/indra/llmessage/message.cpp @@ -272,6 +272,9 @@ LLMessageSystem::LLMessageSystem(const std::string& filename, U32 port, mDecodedQueue = std::make_shared>>(); mReliableAckQueue = std::make_shared>(); mReceiverThread = std::make_unique(mSocket, mIncomingQueue); + // The receiver thread uses plain LLCircuitData pointers, so circuit + // removal has to go through the graveyard while it is alive. + mCircuitInfo.setDeferredDeletion(true); } // LL_DEBUGS("Messaging") << << "*** port: " << mPort << LL_ENDL; @@ -369,6 +372,9 @@ LLMessageSystem::~LLMessageSystem() // This may wait for the thread to finish. mReceiverThread->shutdown(); } + // The receiver thread is gone, circuits can be deleted right away again. + mCircuitInfo.setDeferredDeletion(false); + mCircuitInfo.cleanupGraveyard(); mMessageTemplates.clear(); // don't delete templates. for_each(mMessageNumbers.begin(), mMessageNumbers.end(), DeletePairedPointer()); @@ -499,6 +505,10 @@ std::unique_ptr LLMessageSystem::decodeDataOwned() S32 compressed_size = zeroCodeExpand(&buffer, &receive_size); LLHost host = pkt.getHost(); + // This runs on the UDP receiver thread, where a circuit removed by the + // main thread is not deleted right away but parked in the circuit + // graveyard (see LLCircuit::removeCircuitData()), so the pointer stays + // valid until this thread is done with the packet. LLCircuitData* cdp = mCircuitInfo.findCircuit(host); const TPACKETID recv_packet_id = ntohl(*((U32*)(&buffer[1]))); const bool recv_reliable = (buffer[0] & LL_RELIABLE_FLAG) != 0; @@ -524,7 +534,7 @@ std::unique_ptr LLMessageSystem::decodeDataOwned() if (!cdp && (mTemplateMessageReader->getMessageName() != _PREHASH_UseCircuitCode)) { - logMsgFromInvalidCircuit(host, pkt.getPacketIDChecked()); + logMsgFromInvalidCircuit(host, recv_reliable); return nullptr; } @@ -1207,17 +1217,25 @@ S32 LLMessageSystem::drainUdpSocket() bool LLMessageSystem::computeDrop() { - bool drop = (mDropPercentage > 0.0f && (ll_frand(100.f) < mDropPercentage)); - if (drop) + // Percentage-based drops don't consume the explicitly requested drop + // count: the original implementation incremented mPacketsToDrop and then + // immediately decremented it again, leaving the count unchanged. + F32 drop_percentage = mDropPercentage.load(std::memory_order_relaxed); + if (drop_percentage > 0.0f && (ll_frand(100.f) < drop_percentage)) { - ++mPacketsToDrop; + return true; } - if (mPacketsToDrop > 0) + + // Consume one of the drops requested by dropPackets(), if any are left. + U32 packets_to_drop = mPacketsToDrop.load(std::memory_order_relaxed); + while (packets_to_drop > 0 + && !mPacketsToDrop.compare_exchange_weak(packets_to_drop, + packets_to_drop - 1, + std::memory_order_relaxed)) { - --mPacketsToDrop; - drop = true; + // 'packets_to_drop' has been updated with the current value, retry. } - return drop; + return packets_to_drop > 0; } bool LLMessageSystem::isHighPriorityMessage(const LLPacketBuffer& pkt) const @@ -1258,12 +1276,12 @@ bool LLMessageSystem::isHighPriorityMessage(const LLPacketBuffer& pkt) const void LLMessageSystem::dropPackets(U32 num_to_drop) { - mPacketsToDrop += num_to_drop; + mPacketsToDrop.fetch_add(num_to_drop, std::memory_order_relaxed); } void LLMessageSystem::setDropPercentage(F32 percent_to_drop) { - mDropPercentage = percent_to_drop; + mDropPercentage.store(percent_to_drop, std::memory_order_relaxed); } S32 LLMessageSystem::receivePacketOrDrop(char* datap, bool& packet_id_already_checked) @@ -1405,7 +1423,7 @@ void LLMessageSystem::dumpPacketRingStats() << "buffered_bytes=" << (mHighPriorityInbound.getNumBufferedBytes() + mLowPriorityInbound.getNumBufferedBytes()) << "recently_dropped=" << mNumDroppedPackets << "total_dropped=" << mNumDroppedPacketsTotal - << "dropped_percentage=" << mDropPercentage << "%" + << "dropped_percentage=" << mDropPercentage.load(std::memory_order_relaxed) << "%" << "bytes_IN=" << mActualBytesIn << "bytes_OUT=" << mActualBytesOut << LL_ENDL; mNumDroppedPackets = 0; @@ -3277,10 +3295,12 @@ void LLMessageSystem::summarizeLogs(std::ostream& str) iter != end; iter++) { const LLMessageTemplate* mt = iter->second; - if(mt->mTotalDecoded > 0) + U32 total_decoded = mt->getTotalDecoded(); + if(total_decoded > 0) { - avg = mt->mTotalDecodeTime / (F32)mt->mTotalDecoded; - buffer = llformat( "%35s%10u%10f%10f%10f", mt->mName, mt->mTotalDecoded, mt->mTotalDecodeTime, mt->mMaxDecodeTimePerMsg, avg); + F32 total_decode_time = mt->getTotalDecodeTime(); + avg = total_decode_time / (F32)total_decoded; + buffer = llformat( "%35s%10u%10f%10f%10f", mt->mName, total_decoded, total_decode_time, mt->getMaxDecodeTimePerMsg(), avg); str << buffer << std::endl; } } @@ -3326,9 +3346,7 @@ void LLMessageSystem::dumpReceiveCounts() iter != end; iter++) { LLMessageTemplate* mt = iter->second; - mt->mReceiveCount = 0; - mt->mReceiveBytes = 0; - mt->mReceiveInvalid = 0; + mt->resetReceiveCounts(); } S32 i; @@ -3337,12 +3355,8 @@ void LLMessageSystem::dumpReceiveCounts() mt = get_ptr_in_map(mMessageNumbers,mMessageCountList[i].mMessageNum); if (mt) { - mt->mReceiveCount++; - mt->mReceiveBytes += mMessageCountList[i].mMessageBytes; - if (mMessageCountList[i].mInvalid) - { - mt->mReceiveInvalid++; - } + mt->recordReceiveCount((U32)mMessageCountList[i].mMessageBytes, + mMessageCountList[i].mInvalid); } } @@ -3354,10 +3368,10 @@ void LLMessageSystem::dumpReceiveCounts() iter != end; iter++) { const LLMessageTemplate* mt = iter->second; - if (mt->mReceiveCount > 0) + if (mt->getReceiveCount() > 0) { - LL_INFOS("Messaging") << "Num: " << std::setw(3) << mt->mReceiveCount << " Bytes: " << std::setw(6) << mt->mReceiveBytes - << " Invalid: " << std::setw(3) << mt->mReceiveInvalid << LL_ENDL; + LL_INFOS("Messaging") << "Num: " << std::setw(3) << mt->getReceiveCount() << " Bytes: " << std::setw(6) << mt->getReceiveBytes() + << " Invalid: " << std::setw(3) << mt->getReceiveInvalid() << LL_ENDL; } } } @@ -3573,19 +3587,14 @@ namespace if (LLMessageReader::getTimeDecodes()) { - msg_template->mTotalDecoded++; - msg_template->mTotalDecodeTime += decode_time; - - if (msg_template->mMaxDecodeTimePerMsg < decode_time) - { - msg_template->mMaxDecodeTimePerMsg = decode_time; - } + msg_template->recordDecodeTime(decode_time); if (decode_time > LLMessageReader::getTimeDecodesSpamThreshold()) { + U32 total_decoded = msg_template->getTotalDecoded(); LL_DEBUGS() << "--------- Message " << msg_template->mName << " decode took " << decode_time << " seconds. (" - << msg_template->mMaxDecodeTimePerMsg << " max, " - << (msg_template->mTotalDecodeTime / msg_template->mTotalDecoded) << " avg)" << LL_ENDL; + << msg_template->getMaxDecodeTimePerMsg() << " max, " + << (msg_template->getTotalDecodeTime() / (F32)llmax(total_decoded, (U32)1)) << " avg)" << LL_ENDL; } } } diff --git a/indra/llmessage/message.h b/indra/llmessage/message.h index 3b8181ed66..7984a9208a 100644 --- a/indra/llmessage/message.h +++ b/indra/llmessage/message.h @@ -27,6 +27,7 @@ #ifndef LL_MESSAGE_H #define LL_MESSAGE_H +#include #include #include #include @@ -1052,8 +1053,10 @@ class LLMessageSystem : public LLMessageSenderInterface // Packet-loss simulation and byte-accounting state std::atomic mActualBytesIn; S32 mActualBytesOut; - F32 mDropPercentage; // % of inbound packets to drop - U32 mPacketsToDrop; // drop next N inbound packets + // computeDrop() runs on LLUDPReceiverThread while setDropPercentage() + // and dropPackets() are driven from the main thread, so both are atomic. + std::atomic mDropPercentage; // % of inbound packets to drop + std::atomic mPacketsToDrop; // drop next N inbound packets S32 mNumDroppedPackets; // inbound S32 mNumDroppedPacketsTotal;// inbound diff --git a/indra/llmessage/tests/llmessagetemplate_test.cpp b/indra/llmessage/tests/llmessagetemplate_test.cpp new file mode 100644 index 0000000000..587e5ae853 --- /dev/null +++ b/indra/llmessage/tests/llmessagetemplate_test.cpp @@ -0,0 +1,144 @@ +/** + * @file llmessagetemplate_test.cpp + * @brief LLMessageTemplate counter and timing statistics tests. + * + * $LicenseInfo:firstyear=2026&license=viewerlgpl$ + * Second Life Viewer Source Code + * Copyright (C) 2026, Linden Research, Inc. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; + * version 2.1 of the License only. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + * + * Linden Research, Inc., 945 Battery Street, San Francisco, CA 94111 USA + * $/LicenseInfo$ + */ + +#include "linden_common.h" +#include "lltut.h" +#include "llmessagetemplate.h" + +#include +#include + +namespace tut +{ + struct messagetemplate + { + LLMessageTemplate mTemplate{ "TestMessage", (U32)1, MFT_HIGH }; + }; + typedef test_group messagetemplate_t; + typedef messagetemplate_t::object messagetemplate_object_t; + tut::messagetemplate_t tut_messagetemplate("LLMessageTemplate"); + + // Receive counters start at zero and accumulate. + template<> template<> + void messagetemplate_object_t::test<1>() + { + ensure_equals("initial receive count", mTemplate.getReceiveCount(), (U32)0); + ensure_equals("initial receive bytes", mTemplate.getReceiveBytes(), (U32)0); + ensure_equals("initial invalid count", mTemplate.getReceiveInvalid(), (U32)0); + + mTemplate.recordReceive(); + ensure_equals("receive count", mTemplate.getReceiveCount(), (U32)1); + + mTemplate.recordReceiveCount(100, false); + mTemplate.recordReceiveCount(50, true); + ensure_equals("receive count", mTemplate.getReceiveCount(), (U32)3); + ensure_equals("receive bytes", mTemplate.getReceiveBytes(), (U32)150); + ensure_equals("invalid count", mTemplate.getReceiveInvalid(), (U32)1); + + mTemplate.resetReceiveCounts(); + ensure_equals("reset receive count", mTemplate.getReceiveCount(), (U32)0); + ensure_equals("reset receive bytes", mTemplate.getReceiveBytes(), (U32)0); + ensure_equals("reset invalid count", mTemplate.getReceiveInvalid(), (U32)0); + } + + // Decode timing statistics accumulate and track the maximum. + template<> template<> + void messagetemplate_object_t::test<2>() + { + ensure_equals("initial decoded", mTemplate.getTotalDecoded(), (U32)0); + ensure_equals("initial decode time", mTemplate.getTotalDecodeTime(), 0.f); + ensure_equals("initial max decode time", mTemplate.getMaxDecodeTimePerMsg(), 0.f); + + mTemplate.recordDecodeTime(1.f); + mTemplate.recordDecodeTime(3.f); + mTemplate.recordDecodeTime(2.f); + + ensure_equals("decoded", mTemplate.getTotalDecoded(), (U32)3); + ensure_equals("total decode time", mTemplate.getTotalDecodeTime(), 6.f); + ensure_equals("max decode time", mTemplate.getMaxDecodeTimePerMsg(), 3.f); + + mTemplate.resetDecodeStats(); + ensure_equals("reset decoded", mTemplate.getTotalDecoded(), (U32)0); + ensure_equals("reset decode time", mTemplate.getTotalDecodeTime(), 0.f); + ensure_equals("reset max decode time", mTemplate.getMaxDecodeTimePerMsg(), 0.f); + } + + // Counters are updated by the UDP receiver thread while the main thread + // reports on them: concurrent updates must not lose any counts. + template<> template<> + void messagetemplate_object_t::test<3>() + { + const int thread_count = 4; + const int per_thread = 1000; + + std::vector threads; + for (int i = 0; i < thread_count; ++i) + { + threads.emplace_back([this]() + { + for (int j = 0; j < per_thread; ++j) + { + mTemplate.recordReceiveCount(10, (j % 2) == 0); + mTemplate.recordDecodeTime(1.f); + } + }); + } + for (auto& thread : threads) + { + thread.join(); + } + + ensure_equals("receive count", mTemplate.getReceiveCount(), (U32)(thread_count * per_thread)); + ensure_equals("receive bytes", mTemplate.getReceiveBytes(), (U32)(thread_count * per_thread * 10)); + ensure_equals("invalid count", mTemplate.getReceiveInvalid(), (U32)(thread_count * per_thread / 2)); + ensure_equals("decoded", mTemplate.getTotalDecoded(), (U32)(thread_count * per_thread)); + ensure_equals("total decode time", mTemplate.getTotalDecodeTime(), (F32)(thread_count * per_thread)); + ensure_equals("max decode time", mTemplate.getMaxDecodeTimePerMsg(), 1.f); + } + + // Copying a template carries the counters over, so that the explicit copy + // operations stay in sync with the atomic members. + template<> template<> + void messagetemplate_object_t::test<4>() + { + mTemplate.recordReceiveCount(42, true); + mTemplate.recordDecodeTime(0.5f); + + LLMessageTemplate copied(mTemplate); + ensure_equals("copied receive count", copied.getReceiveCount(), (U32)1); + ensure_equals("copied receive bytes", copied.getReceiveBytes(), (U32)42); + ensure_equals("copied invalid count", copied.getReceiveInvalid(), (U32)1); + ensure_equals("copied decoded", copied.getTotalDecoded(), (U32)1); + ensure_equals("copied decode time", copied.getTotalDecodeTime(), 0.5f); + ensure_equals("copied max decode time", copied.getMaxDecodeTimePerMsg(), 0.5f); + + LLMessageTemplate assigned("OtherTestMessage", (U32)2, MFT_HIGH); + assigned = mTemplate; + ensure_equals("assigned receive count", assigned.getReceiveCount(), (U32)1); + ensure_equals("assigned decoded", assigned.getTotalDecoded(), (U32)1); + ensure_equals("assigned name", std::string(assigned.mName), std::string("TestMessage")); + } +} diff --git a/indra/newview/llfloaterimsession.cpp b/indra/newview/llfloaterimsession.cpp index 73bce6f2fd..57ce85f3d6 100644 --- a/indra/newview/llfloaterimsession.cpp +++ b/indra/newview/llfloaterimsession.cpp @@ -310,6 +310,8 @@ void LLFloaterIMSession::initIMSession(const LLUUID& session_id) mSessionID = session_id; mSession = LLIMModel::getInstance()->findIMSession(mSessionID); + updateUniqueName(); + if (mSession) { mIsP2PChat = mSession->isP2PSessionType(); diff --git a/indra/newview/llfloaterimsessiontab.cpp b/indra/newview/llfloaterimsessiontab.cpp index b70d21e2c5..b254db1678 100644 --- a/indra/newview/llfloaterimsessiontab.cpp +++ b/indra/newview/llfloaterimsessiontab.cpp @@ -328,6 +328,8 @@ bool LLFloaterIMSessionTab::postBuild() setOpenPositioning(LLFloaterEnums::POSITIONING_RELATIVE); + updateUniqueName(); + mSaveRect = isNearbyChat() && !gSavedPerAccountSettings.getBOOL("NearbyChatIsNotTornOff"); initRectControl(); @@ -393,6 +395,22 @@ LLParticipantList* LLFloaterIMSessionTab::getParticipantList() return dynamic_cast(LLFloaterIMContainer::getInstance()->getSessionModel(mSessionID)); } +void LLFloaterIMSessionTab::updateUniqueName() +{ + if (isNearbyChat()) + { + return; + } + + LLUUID name_id = mSessionID; + if (mSession && mSession->isP2PSessionType()) + { + name_id = mSession->mOtherParticipantID; + } + + setName("panel_im_" + name_id.asString()); +} + // virtual void LLFloaterIMSessionTab::draw() { diff --git a/indra/newview/llfloaterimsessiontab.h b/indra/newview/llfloaterimsessiontab.h index d97b0eaa33..be9b3fe5a2 100644 --- a/indra/newview/llfloaterimsessiontab.h +++ b/indra/newview/llfloaterimsessiontab.h @@ -146,6 +146,9 @@ class LLFloaterIMSessionTab // Update the input field help text and other places that need the session name virtual void updateSessionName(const std::string& name); + // Use a unique root name for LEAP conversation lookups. + void updateUniqueName(); + // set the enable/disable state for the Call button virtual void enableDisableCallBtn(); diff --git a/indra/newview/lltoast.cpp b/indra/newview/lltoast.cpp index 3c939a88e5..5a79953879 100644 --- a/indra/newview/lltoast.cpp +++ b/indra/newview/lltoast.cpp @@ -30,6 +30,7 @@ #include "llbutton.h" #include "llfocusmgr.h" +#include "llimview.h" #include "llnotifications.h" #include "llviewercontrol.h" @@ -167,9 +168,36 @@ bool LLToast::postBuild() mTimer->stop(); } + updateUniqueName(); + return true; } +//-------------------------------------------------------------------------- +void LLToast::updateUniqueName() +{ + // Several toasts can be alive at once for the same session/group, so unique notification id is needed. + // Session/group id provides a filterable name prefix for LEAP. + LLUUID context_id = mSessionID; + if (mSessionID.notNull()) + { + LLIMModel::LLIMSession* session = LLIMModel::getInstance()->findIMSession(mSessionID); + if (session && session->isP2PSessionType()) + { + context_id = session->mOtherParticipantID; + } + } + + std::string name = "toast_"; + if (context_id.notNull()) + { + name += context_id.asString() + "_"; + } + name += mNotificationID.asString(); + + setName(name); +} + //-------------------------------------------------------------------------- void LLToast::setHideButtonEnabled(bool enabled) { diff --git a/indra/newview/lltoast.h b/indra/newview/lltoast.h index a73f7820ba..be85df66d8 100644 --- a/indra/newview/lltoast.h +++ b/indra/newview/lltoast.h @@ -214,6 +214,9 @@ class LLToast : public LLModalDialog, public LLInstanceTracker void setFading(bool fading); + // Use a unique root name for LEAP toast lookups. + void updateUniqueName(); + LLUUID mNotificationID; LLUUID mSessionID; LLNotificationPtr mNotification;