diff --git a/doc/TESTING.md b/doc/TESTING.md index 71bff29698..8d5afc7d00 100644 --- a/doc/TESTING.md +++ b/doc/TESTING.md @@ -63,6 +63,7 @@ Everything a test file defines goes in an anonymous namespace. A test binary is - `VertexBufferProbe` (`llrender/llfonttextcache.h`); - `RenderProbe` (`llrender/llrender.h`); - `VolumeProbe` (`llmath/llvolume.h`); + - `CircuitProbe` (`llmessage/llcircuit.h`); - `TextViewProbe` (`llui/text/altextview.h` and `altextruler.h`); - `alincludesearch_data` (`alscript/preprocessor/alincludesearch.h`). - **Internal is not enough for LL's name-keyed registries.** `LLSingleton`, `LLInstanceTracker` and `LLThreadLocalSingletonPointer` key on `typeid(T).name()`, which is `(anonymous namespace)::X` for every anonymous `X`. A type used with one of them needs a name unique across the tree. diff --git a/indra/llmessage/CMakeLists.txt b/indra/llmessage/CMakeLists.txt index 1fafe5240d..7db102668e 100644 --- a/indra/llmessage/CMakeLists.txt +++ b/indra/llmessage/CMakeLists.txt @@ -170,9 +170,9 @@ target_precompile_headers(llmessage REUSE_FROM llprecompiled) # tests if(AL_BUILD_TESTS) - # These three stand in for parts of llmessage and llcorehttp with stubs of + # These four stand in for parts of llmessage and llcorehttp with stubs of # their own, so they build their one file and do not link llmessage. - foreach(test llcoproceduremanager lltrustedmessageservice lltemplatemessagedispatcher) + foreach(test llcoproceduremanager lltrustedmessageservice lltemplatemessagedispatcher llcircuit) al_add_test_binary( ${test}_test PROJECT llmessage diff --git a/indra/llmessage/llcircuit.cpp b/indra/llmessage/llcircuit.cpp index 2c97f0cf02..cac0529fb8 100644 --- a/indra/llmessage/llcircuit.cpp +++ b/indra/llmessage/llcircuit.cpp @@ -63,6 +63,8 @@ const S32 PING_RELEASE_BLOCK = 2; // How many pings behind we have to be to co const F32Seconds TARGET_PERIOD_LENGTH(5.f); const F32Seconds LL_DUPLICATE_SUPPRESSION_TIMEOUT(60.f); //this can be long, as time-based cleanup is // only done when wrapping packetids, now... +const F32Seconds LL_DUPLICATE_COPY_WINDOW(5.f); // how long after a reliable packet, or its latest copy, a copy may still arrive: + // an unmarked copy counts as a duplicate, and a ping keeps its number LLCircuitData::LLCircuitData(const LLHost &host, TPACKETID in_id, const F32Seconds circuit_heartbeat_interval, const F32Seconds circuit_timeout) @@ -598,9 +600,30 @@ void LLCircuit::resendUnackedPackets(S32& unacked_list_length, S32& unacked_list } -bool LLCircuitData::isDuplicateResend(TPACKETID packetnum) +bool LLCircuitData::checkDuplicate(TPACKETID packetnum, bool resent) { - return (mRecentlyReceivedReliablePackets.find(packetnum) != mRecentlyReceivedReliablePackets.end()); + packet_time_map::iterator it = mRecentlyReceivedReliablePackets.find(packetnum); + if (it == mRecentlyReceivedReliablePackets.end()) + { + return false; + } + + // A copy the network made of a packet comes within moments of the last, + // unmarked. An older number an unmarked packet meets is left from + // numbering the sender has since started over, on a circuit kept up. + U64Microseconds mt_usec = LLMessageSystem::getMessageTimeUsecs(); + U64Microseconds delta_t_usec = mt_usec - it->second; + F64Seconds delta_t_sec = delta_t_usec; + if (!resent && delta_t_sec >= LL_DUPLICATE_COPY_WINDOW) + { + return false; + } + + // While acks for a packet are lost its sender goes on resending it, and + // the last resend can come after the ping that says it was acked: the + // window a copy may still arrive in runs from the latest copy. + it->second = mt_usec; + return true; } @@ -858,11 +881,10 @@ bool LLCircuitData::updateWatchDogTimers(LLMessageSystem *msgsys) } // WARNING! - // Duplicate suppression can FAIL if packets are delivered out of - // order, although it's EXTREMELY unlikely. It would require - // that the ping get delivered out of order enough that the ACK - // for the packet that it was out of order with was received BEFORE - // the ping was sent. + // Duplicate suppression can FAIL if a copy of a packet is delivered + // after the ping, and longer than LL_DUPLICATE_COPY_WINDOW after the + // packet or its last copy came: the ping lets the receiver forget a + // number its oldest unacked passes once nothing of it came for that long. // Find the current oldest reliable packetID // This is to handle the case if we actually manage to wrap our @@ -999,26 +1021,40 @@ void LLCircuitData::clearDuplicateList(TPACKETID oldest_id) { // purge old data from the duplicate suppression queue - // we want to KEEP all x where oldest_id <= x <= last incoming packet, and delete everything else. + // we want to KEEP all x where oldest_id <= x <= last incoming packet, and delete everything else + // once no copy of it can still be coming. //LL_INFOS() << mHost << ": clearing before oldest " << oldest_id << LL_ENDL; //LL_INFOS() << "Recent list before: " << mRecentlyReceivedReliablePackets.size() << LL_ENDL; + U64Microseconds mt_usec = LLMessageSystem::getMessageTimeUsecs(); + + packet_time_map::iterator pit; if (oldest_id < mHighestPacketID) { - // Clean up everything with a packet ID less than oldest_id. - packet_time_map::iterator pit_start; - packet_time_map::iterator pit_end; - pit_start = mRecentlyReceivedReliablePackets.begin(); - pit_end = mRecentlyReceivedReliablePackets.lower_bound(oldest_id); - mRecentlyReceivedReliablePackets.erase(pit_start, pit_end); + // Clean up everything with a packet ID less than oldest_id that was + // handled, or last seen again, longer ago than a copy of it takes to + // arrive. The sender + // pings once it has the ack, and a resend that crossed the ack, or a + // copy the network made, can come after the ping. + packet_time_map::iterator pit_end = mRecentlyReceivedReliablePackets.lower_bound(oldest_id); + for (pit = mRecentlyReceivedReliablePackets.begin(); pit != pit_end; ) + { + U64Microseconds delta_t_usec = mt_usec - (*pit).second; + F64Seconds delta_t_sec = delta_t_usec; + if (delta_t_sec >= LL_DUPLICATE_COPY_WINDOW) + { + mRecentlyReceivedReliablePackets.erase(pit++); + } + else + { + ++pit; + } + } } // Do timeout checks on everything with an ID > mHighestPacketID. // This should be empty except for wrapping IDs. Thus, this should be // highly rare. - U64Microseconds mt_usec = LLMessageSystem::getMessageTimeUsecs(); - - packet_time_map::iterator pit; for(pit = mRecentlyReceivedReliablePackets.upper_bound(mHighestPacketID); pit != mRecentlyReceivedReliablePackets.end(); ) { @@ -1076,13 +1112,13 @@ bool LLCircuitData::collectRAck(TPACKETID packet_num) { // 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); - if (mAckCreationTime == 0) - { + // and the acks have waited since now, whether the last ones went in + // a PacketAck or on the end of a packet sent to the host. mAckCreationTime = getAgeInSeconds(); } + + mAcks.push_back(packet_num); return true; } @@ -1136,7 +1172,6 @@ void LLCircuit::sendAcks(F32 collect_time) // empty out the acks list cd->mAcks.clear(); - cd->mAckCreationTime = 0.f; } // remove data map mSendAckMap.erase(cur_it); @@ -1289,9 +1324,13 @@ void LLCircuitData::pingTimerStop(const U8 ping_id) } U32Milliseconds msec = delta_ping*mHeartbeatInterval + time; - setPingDelay(msec); + // Only the pings sent after this one are still in flight. Count them + // before averaging: setPingDelay starts from getPingDelayAveraged(), + // which takes the time in flight of every ping counted as a floor. mPingsInTransit = delta_ping; + setPingDelay(msec); + if (mBlocked && (mPingsInTransit <= PING_RELEASE_BLOCK)) { mBlocked = false; diff --git a/indra/llmessage/llcircuit.h b/indra/llmessage/llcircuit.h index 1f43e4d272..8ea96652ad 100644 --- a/indra/llmessage/llcircuit.h +++ b/indra/llmessage/llcircuit.h @@ -69,6 +69,11 @@ class LLMessageSystem; class LLEncodedDatagramService; class LLSD; +namespace ll_test +{ + struct CircuitProbe; +} + // // Classes // @@ -160,6 +165,7 @@ class LLCircuitData friend class LLCircuit; friend class LLMessageSystem; friend class LLEncodedDatagramService; + friend struct ll_test::CircuitProbe; friend void crash_on_spaceserver_timeout (const LLHost &host, void *); // HACK, so it has access to setAlive() so it can send a final shutdown message. protected: TPACKETID nextPacketOutID(); @@ -176,7 +182,11 @@ class LLCircuitData bool updateWatchDogTimers(LLMessageSystem *msgsys); // Return false if the circuit is dead and should be cleaned up void addReliablePacket(S32 mSocket, U8 *buf_ptr, S32 buf_len, LLReliablePacketParams *params); - bool isDuplicateResend(TPACKETID packetnum); + // Whether a reliable packet numbered packetnum was handled already: any + // number the record holds for a resent packet, and for an unmarked one + // only a number handled or seen again moments ago. A duplicate's number + // is kept as seen now. + bool checkDuplicate(TPACKETID packetnum, bool resent); // 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 diff --git a/indra/llmessage/llzerocode.h b/indra/llmessage/llzerocode.h index ac014dc550..53a886ed42 100644 --- a/indra/llmessage/llzerocode.h +++ b/indra/llmessage/llzerocode.h @@ -35,12 +35,12 @@ // first header_size bytes of the buffer (the packet header - flags, // sequence number, offset, etc.) untouched aside from the flag bit below. // -// Runs of zero bytes in the body are replaced by a two-byte token: -// 0x00 N - represents N zero bytes, for N in 1..254 -// 0x00 0x00 N - represents (255 + N) zero bytes (wrap/overflow case, -// produced by decode()'s wire format but never emitted -// by encode(), which instead starts a fresh 0x00 token -// every 255 zero bytes) +// Runs of zero bytes in the body are replaced by a 0x00 byte and a count: +// 0x00 N - represents N zero bytes, for N in 1..255; encode() +// starts a fresh token every 255 zero bytes +// 0x00 0x00 N - represents (256 + N) zero bytes, each further 0x00 +// before the count adding 256 (read by decode(), never +// written by encode()) namespace LLZeroCode { // High bit of the first header byte: set by encode() and cleared by @@ -191,7 +191,9 @@ namespace LLZeroCode } else { - if ((U32)(outptr - dst) + (*inptr) > dst_capacity) + // The run's first zero is already written, so it has + // the count less one to go. + if ((U32)(outptr - dst) + (*inptr) - 1 > dst_capacity) { // Discard rather than decode on into a scrambled // buffer, the same as the two cases above. diff --git a/indra/llmessage/message.cpp b/indra/llmessage/message.cpp index 81d1bbfcac..e2cab409c5 100644 --- a/indra/llmessage/message.cpp +++ b/indra/llmessage/message.cpp @@ -639,46 +639,36 @@ bool LLMessageSystem::checkMessages(LockMessageChecker&, S64 frame_count, if (buffer[0] & LL_RESENT_FLAG) { recv_resent = true; - if (cdp && cdp->isDuplicateResend(mCurrentRecvPacketID)) + } + + // A reliable packet the network delivers twice comes the second + // time without the resent flag, and is as much a duplicate as a + // resend. Only reliable packets' numbers are recorded, so an + // unreliable one is not looked up. + if (recv_reliable && cdp && cdp->checkDuplicate(mCurrentRecvPacketID, recv_resent)) + { + // We need to ACK here to suppress + // further resends of packets we've + // already seen. + cdp->collectRAck(mCurrentRecvPacketID); + + LL_DEBUGS("Messaging") << "Discarding duplicate from " << host << LL_ENDL; + if(mVerboseLog) { - // We need to ACK here to suppress - // further resends of packets we've - // already seen. - if (recv_reliable) - { - //mAckList.addData(new LLPacketAck(host, mCurrentRecvPacketID)); - // *************************************** - // TESTING CODE - //if(mCircuitInfo.mCurrentCircuit->mHost != host) - //{ - // LL_WARNS("Messaging") << "DISCARDED PACKET HOST MISMATCH! HOST: " - // << host << " CIRCUIT: " - // << mCircuitInfo.mCurrentCircuit->mHost - // << LL_ENDL; - //} - // *************************************** - //mCircuitInfo.mCurrentCircuit->mAcks.put(mCurrentRecvPacketID); - cdp->collectRAck(mCurrentRecvPacketID); - } - - LL_DEBUGS("Messaging") << "Discarding duplicate resend from " << host << LL_ENDL; - if(mVerboseLog) - { - std::ostringstream str; - str << "MSG: <- " << host; - std::string tbuf; - tbuf = llformat( "\t%6d\t%6d\t%6d ", receive_size, (mIncomingCompressedSize ? mIncomingCompressedSize : receive_size), mCurrentRecvPacketID); - str << tbuf << "(unknown)" - << (recv_reliable ? " reliable" : "") - << " resent " - << ((num_acks > 0) ? "acks" : "") - << " DISCARD DUPLICATE"; - LL_INFOS("Messaging") << str.str() << LL_ENDL; - } - mPacketsIn++; - valid_packet = false; - continue; + std::ostringstream str; + str << "MSG: <- " << host; + std::string tbuf; + tbuf = llformat( "\t%6d\t%6d\t%6d ", receive_size, (mIncomingCompressedSize ? mIncomingCompressedSize : receive_size), mCurrentRecvPacketID); + str << tbuf << "(unknown)" + << " reliable" + << (recv_resent ? " resent " : " ") + << ((num_acks > 0) ? "acks" : "") + << " DISCARD DUPLICATE"; + LL_INFOS("Messaging") << str.str() << LL_ENDL; } + mPacketsIn++; + valid_packet = false; + continue; } // UseCircuitCode can be a valid, off-circuit packet. @@ -1168,10 +1158,12 @@ S32 LLMessageSystem::bufferInboundPacket() // Check packet sequencing here, in true socket-arrival order, before // this packet is sorted into the high/low priority inbound queue. - // Skip genuine duplicate resends, same as checkMessages()/logValidMsg() - // would do further downstream. + // Skip duplicates of reliable packets already handled, marked resent + // or not, same as checkMessages()/logValidMsg() would do further + // downstream. bool recv_resent = (data[0] & LL_RESENT_FLAG) != 0; - if (!recv_resent || !cdp->isDuplicateResend(recv_packet_id)) + bool recv_reliable = (data[0] & LL_RELIABLE_FLAG) != 0; + if (!recv_reliable || !cdp->checkDuplicate(recv_packet_id, recv_resent)) { cdp->checkPacketInID(recv_packet_id, recv_resent); pkt.setPacketIDChecked(true); @@ -1651,9 +1643,10 @@ S32 LLMessageSystem::sendMessage(const LLHost &host) mReliablePacketsOut++; } - // tack packet acks onto the end of this message; signed math so an - // over-MTU payload yields a negative count instead of wrapping huge - S32 space_left = (MTUBYTES - (S32)buffer_length) / (S32)sizeof(TPACKETID); // space left for packet ids + // tack packet acks onto the end of this message, and the byte that counts + // them; signed math so an over-MTU payload yields a negative count + // instead of wrapping huge + S32 space_left = (MTUBYTES - (S32)buffer_length - 1) / (S32)sizeof(TPACKETID); // space left for packet ids S32 ack_count = (S32)cdp->mAcks.size(); bool is_ack_appended = false; std::vector acks; @@ -2010,16 +2003,11 @@ void LLMessageSystem::disableCircuit(const LLHost &host) LL_INFOS("Messaging") << "Host " << LLHost(old_ip, old_port) << " circuit " << code << " removed from lookup table" << LL_ENDL; gMessageSystem->mIPPortToCircuitCode.erase(ip_port); } - mCircuitInfo.removeCircuitData(host); - } - else - { - // Sigh, since we can open circuits which don't have circuit - // codes, it's possible for this to happen... - - LL_WARNS("Messaging") << "Couldn't find circuit code for " << host << LL_ENDL; } + // A circuit opened without a code, as the viewer opens every circuit to + // a simulator, goes too. + mCircuitInfo.removeCircuitData(host); } diff --git a/indra/llmessage/message.h b/indra/llmessage/message.h index 3b3b85aaae..5a927d5c6a 100644 --- a/indra/llmessage/message.h +++ b/indra/llmessage/message.h @@ -157,10 +157,11 @@ class LLMessageStringTable : public LLSingleton // // Zero-coding (applied when LL_ZERO_CODE_FLAG is set): // -// Runs of zero bytes in the body are replaced by a two-byte token: -// 0x00 N – represents (N + 1) zero bytes, for N in 1..254 -// 0x00 0x00 N – represents (256 + N) zero bytes (wrap/overflow case) -// A literal 0x00 byte that starts no run is encoded as 0x00 0x00 0x00. +// Runs of zero bytes in the body are replaced by a 0x00 byte and a count: +// 0x00 N – represents N zero bytes, for N in 1..255 +// 0x00 0x00 N – represents (256 + N) zero bytes, each further 0x00 +// before the count adding 256 (read, never written) +// A lone zero byte is a run of one, 0x00 0x01. // The six-byte packet header is excluded from zero-coding and is always // transmitted as-is. diff --git a/indra/llmessage/tests/llcircuit_test.cpp b/indra/llmessage/tests/llcircuit_test.cpp new file mode 100644 index 0000000000..ed8f6cb68d --- /dev/null +++ b/indra/llmessage/tests/llcircuit_test.cpp @@ -0,0 +1,455 @@ +/** + * @file llcircuit_test.cpp + * @brief LLCircuitData's averaged ping, and the acks it collects. + * + * $LicenseInfo:firstyear=2026&license=viewerlgpl$ + * Alchemy Viewer Source Code + * Copyright (C) 2026, Rye + * + * 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 + * $/LicenseInfo$ + */ + +#include "linden_common.h" + +#include "llcircuit.h" + +#include "llpacketbuffer.h" +#include "llpacketring.h" +#include "llpounceable.h" +#include "lltransfermanager.h" +#include "message.h" + +#include "altestmatchers.h" + +#include + +#include +#include + +// The circuit reads its times from the message system's clock, which here is +// the test's own, and sends its acks through the message system, which here +// records the IDs each PacketAck carries. +namespace +{ + F64Seconds sMessageTime; + std::vector sAckedIDs; + std::vector sMessageIDs; +} + +F64Seconds LLMessageSystem::getMessageTimeSeconds(const bool) +{ + return sMessageTime; +} + +U64Microseconds LLMessageSystem::getMessageTimeUsecs(const bool) +{ + return sMessageTime; +} + +void LLMessageSystem::newMessageFast(const char*) +{ + sMessageIDs.clear(); +} + +void LLMessageSystem::nextBlockFast(const char*) +{ +} + +void LLMessageSystem::addU8Fast(const char*, U8) +{ +} + +void LLMessageSystem::addU32Fast(const char*, U32 u) +{ + sMessageIDs.push_back(u); +} + +S32 LLMessageSystem::sendMessage(const LLHost&) +{ + sAckedIDs.insert(sAckedIDs.end(), sMessageIDs.begin(), sMessageIDs.end()); + return 0; +} + +// The rest of llmessage that llcircuit.cpp calls. A circuit makes its throttle +// group, and as it goes ends its host's transfers and reads whether the +// message system logs verbosely; the rest is reached only by sending, +// resending or acking a reliable packet, which no case does. The message +// system a circuit reads is made with none of the real one's sockets or +// templates. +LLPounceable gMessageSystem; + +LLMessageSystem::LLMessageSystem(const std::string&, U32, S32, S32, S32, bool, const F32 circuit_heartbeat_interval, + const F32 circuit_timeout) +: mVerboseLog(false), + mCircuitInfo(F32Seconds(circuit_heartbeat_interval), F32Seconds(circuit_timeout)) +{ +} + +LLMessageSystem::~LLMessageSystem() +{ +} + +LLPacketRing::LLPacketRing() +{ +} + +LLPacketRing::~LLPacketRing() +{ +} + +LLPacketBuffer::~LLPacketBuffer() +{ +} + +S32 LLMessageSystem::sendMessage(const LLHost&, LLStoredMessagePtr) +{ + return 0; +} + +bool LLMessageSystem::sendPacketToSocket(const char*, S32, LLHost) +{ + return false; +} + +LLTransferManager gTransferManager; + +LLTransferManager::LLTransferManager() +{ +} + +LLTransferManager::~LLTransferManager() +{ +} + +void LLTransferManager::cleanupConnection(const LLHost&) +{ +} + +LLThrottleGroup::LLThrottleGroup() +{ +} + +bool LLThrottleGroup::checkOverflow(S32, F32) +{ + return false; +} + +bool LLThrottleGroup::throttleOverflow(S32, F32) +{ + return false; +} + +bool LLThrottleGroup::dynamicAdjust() +{ + return false; +} + +LLReliablePacket::LLReliablePacket(S32, U8*, S32, LLReliablePacketParams*) +: mBuffer(nullptr), + mBufferLength(0) +{ +} + +std::string LLHost::getIPandPort() const +{ + return {}; +} + +std::ostream& operator<<(std::ostream& os, const LLHost&) +{ + return os; +} + +char const* const _PREHASH_ID = "ID"; +char const* const _PREHASH_OldestUnacked = "OldestUnacked"; +char const* const _PREHASH_PacketAck = "PacketAck"; +char const* const _PREHASH_Packets = "Packets"; +char const* const _PREHASH_PingID = "PingID"; +char const* const _PREHASH_StartPingCheck = "StartPingCheck"; + +namespace ll_test +{ + // What the message system does with a circuit's acks as packets come in + // from its host and go out to it. + struct CircuitProbe + { + // A reliable packet came in, to be acked. + static void receiveReliable(LLCircuitData& circuit, TPACKETID id) + { + circuit.collectRAck(id); + } + + // A packet went out with every waiting ack appended to it, taken off + // the circuit's list as LLMessageSystem::sendMessage takes them. + static void carryAcks(LLCircuitData& circuit) + { + circuit.mAcks.erase(circuit.mAcks.begin(), circuit.mAcks.end()); + } + + // A reliable packet was handled, its number counted in and recorded + // as LLMessageSystem::checkMessages counts and records it. + static void handled(LLCircuitData& circuit, TPACKETID id) + { + circuit.checkPacketInID(id, false); + circuit.mRecentlyReceivedReliablePackets[id] = LLMessageSystem::getMessageTimeUsecs(); + } + + static bool checkDuplicate(LLCircuitData& circuit, TPACKETID id, bool resent) + { + return circuit.checkDuplicate(id, resent); + } + }; +} + +namespace +{ + using ll_test::CircuitProbe; + using ll_test::WithinRounding; + + // The heartbeat and timeout the message system's tests start it with. + const F32Seconds HEARTBEAT(5.f); + const F32Seconds TIMEOUT(100.f); + + // The clock starts at 1024 s, and every ping starts on a whole second + // below 2048 s: there a double rounds a whole second and 0.2 s up, so an + // answer 0.2 s later measures 200 ms rather than truncating to 199. + const F64Seconds CLOCK_START(1024.0); + + // How long the viewer lets acks gather before it sends them in a + // PacketAck of their own (AckCollectTime). + const F32 ACK_COLLECT_TIME = 0.1f; + + // gMessageSystem is the stub while this lives, and what it was before + // once it ends, with no circuit left on its lists; the clock is at its + // start. + class StubMessageSystemScope + { + public: + StubMessageSystemScope() + : mPrior(gMessageSystem) + { + static LLMessageSystem system("", 0, 0, 0, 0, false, HEARTBEAT.value(), TIMEOUT.value()); + gMessageSystem = &system; + sMessageTime = CLOCK_START; + sAckedIDs.clear(); + } + + ~StubMessageSystemScope() + { + gMessageSystem->mCircuitInfo.mSendAckMap.clear(); + gMessageSystem->mCircuitInfo.mUnackedCircuitMap.clear(); + gMessageSystem = mPrior; + } + + StubMessageSystemScope(const StubMessageSystemScope&) = delete; + StubMessageSystemScope& operator=(const StubMessageSystemScope&) = delete; + + private: + LLMessageSystem* mPrior; + }; + + // What the message system waits before resending a reliable packet on + // the circuit, as it works it out when it sends the packet and again + // when it resends it. + F32Seconds resend_timeout(LLCircuitData& circuit) + { + return llmax(LL_MINIMUM_RELIABLE_TIMEOUT_SECONDS, + F32Seconds(LL_RELIABLE_TIMEOUT_FACTOR * circuit.getPingDelayAveraged())); + } + + // A ping sent at `sent` and answered `round_trip` later, with no other + // ping in flight. The circuit numbers a ping only as it sends one itself, + // so the last one sent is still number 0. + void answer_ping(LLCircuitData& circuit, F64Seconds sent, F64Seconds round_trip) + { + sMessageTime = sent; + circuit.pingTimerStart(); + sMessageTime = sent + round_trip; + circuit.pingTimerStop(0); + } + + // Waits until the circuit is `seconds` older than it was at `since`, by + // its own age, which is what it times its acks by. + void wait_until_aged(const LLCircuitData& circuit, F64 since, F64 seconds) + { + while (circuit.getAgeInSeconds() - since <= seconds) + { + ms_sleep(5); + } + } + + TEST_CASE("llcircuit: an answered ping moves the averaged ping toward its round trip", "[llcircuit]") + { + StubMessageSystemScope message_system; + LLCircuitData circuit(LLHost(), 0, HEARTBEAT, TIMEOUT); + REQUIRE(circuit.getPingDelayAveraged() == F32Milliseconds(INITIAL_PING_VALUE_MSEC)); + + answer_ping(circuit, CLOCK_START, F64Seconds(0.2)); + CHECK(circuit.getPingsInTransit() == 0); + CHECK(circuit.getPingDelay() == U32Milliseconds(200)); + + // A fifth of the way from 1000 ms to 200 ms, and five times that. + CHECK_THAT(circuit.getPingDelayAveraged().value(), WithinRounding(840.f, 2000.f)); + CHECK_THAT(resend_timeout(circuit).value(), WithinRounding(4.2f, 10.f)); + + // Each answer a heartbeat later moves it a fifth of the rest of the way. + const S32 answers = 20; + for (S32 answer = 1; answer < answers; ++answer) + { + answer_ping(circuit, CLOCK_START + answer * HEARTBEAT, F64Seconds(0.2)); + } + const F32 settled = 200.f + 800.f * std::pow(0.8f, F32(answers)); + CHECK_THAT(circuit.getPingDelayAveraged().value(), WithinRounding(settled, 2000.f)); + CHECK_THAT(resend_timeout(circuit).value(), WithinRounding(5.f * settled / 1000.f, 10.f)); + } + + TEST_CASE("llcircuit: a slow answer raises the averaged ping to its round trip at once", "[llcircuit]") + { + StubMessageSystemScope message_system; + LLCircuitData circuit(LLHost(), 0, HEARTBEAT, TIMEOUT); + + answer_ping(circuit, CLOCK_START, F64Seconds(1.5)); + CHECK_THAT(circuit.getPingDelayAveraged().value(), WithinRounding(1500.f, 2000.f)); + CHECK_THAT(resend_timeout(circuit).value(), WithinRounding(7.5f, 10.f)); + + // and falls back a fifth of the way at the next fast one. + answer_ping(circuit, CLOCK_START + HEARTBEAT, F64Seconds(0.2)); + CHECK_THAT(circuit.getPingDelayAveraged().value(), WithinRounding(1240.f, 2000.f)); + } + + TEST_CASE("llcircuit: an ack after a packet carried the last ones waits the collect time", "[llcircuit]") + { + StubMessageSystemScope message_system; + LLCircuitData circuit(LLHost(), 0, HEARTBEAT, TIMEOUT); + LLCircuit& circuits = gMessageSystem->mCircuitInfo; + + // An ack waits, a packet to the host carries it, and the next frame + // finds the circuit with none to send. + const F64 first_ack = circuit.getAgeInSeconds(); + CircuitProbe::receiveReliable(circuit, 1); + CircuitProbe::carryAcks(circuit); + circuits.sendAcks(ACK_COLLECT_TIME); + CHECK(circuits.mSendAckMap.empty()); + CHECK(sAckedIDs.empty()); + + // Well past the collect time after the first, another ack comes in, + // and waits the collect time from its own arrival + wait_until_aged(circuit, first_ack, 2.0 * ACK_COLLECT_TIME); + const F64 second_ack = circuit.getAgeInSeconds(); + CircuitProbe::receiveReliable(circuit, 2); + circuits.sendAcks(ACK_COLLECT_TIME); + CHECK(sAckedIDs.empty()); + CHECK(circuits.mSendAckMap.size() == 1); + + // before it goes in a PacketAck. + wait_until_aged(circuit, second_ack, ACK_COLLECT_TIME); + circuits.sendAcks(ACK_COLLECT_TIME); + CHECK(sAckedIDs == std::vector{ 2 }); + CHECK(circuits.mSendAckMap.empty()); + } + + TEST_CASE("llcircuit: acks gather for the collect time and go in one PacketAck", "[llcircuit]") + { + StubMessageSystemScope message_system; + LLCircuitData circuit(LLHost(), 0, HEARTBEAT, TIMEOUT); + LLCircuit& circuits = gMessageSystem->mCircuitInfo; + + const F64 first_ack = circuit.getAgeInSeconds(); + CircuitProbe::receiveReliable(circuit, 1); + CircuitProbe::receiveReliable(circuit, 2); + circuits.sendAcks(ACK_COLLECT_TIME); + CHECK(sAckedIDs.empty()); + + wait_until_aged(circuit, first_ack, ACK_COLLECT_TIME); + CircuitProbe::receiveReliable(circuit, 3); + circuits.sendAcks(ACK_COLLECT_TIME); + CHECK(sAckedIDs == std::vector{ 1, 2, 3 }); + CHECK(circuits.mSendAckMap.empty()); + } + + TEST_CASE("llcircuit: an unmarked packet is a duplicate only of a number handled moments ago", "[llcircuit]") + { + StubMessageSystemScope message_system; + LLCircuitData circuit(LLHost(), 0, HEARTBEAT, TIMEOUT); + CircuitProbe::handled(circuit, 7); + + // A copy the network made comes unmarked, a resend marked. + CHECK(CircuitProbe::checkDuplicate(circuit, 7, false)); + CHECK(CircuitProbe::checkDuplicate(circuit, 7, true)); + CHECK_FALSE(CircuitProbe::checkDuplicate(circuit, 8, false)); + CHECK_FALSE(CircuitProbe::checkDuplicate(circuit, 8, true)); + + sMessageTime += F64Seconds(4.0); + CHECK(CircuitProbe::checkDuplicate(circuit, 7, false)); + + // Long after, an unmarked packet with that number is a fresh one from + // numbering the sender started over; a resend still matches. + sMessageTime += F64Seconds(6.0); + CHECK_FALSE(CircuitProbe::checkDuplicate(circuit, 7, false)); + CHECK(CircuitProbe::checkDuplicate(circuit, 7, true)); + } + + TEST_CASE("llcircuit: a ping keeps a number handled moments ago that its sender has had acked", "[llcircuit]") + { + StubMessageSystemScope message_system; + LLCircuitData circuit(LLHost(), 0, HEARTBEAT, TIMEOUT); + CircuitProbe::handled(circuit, 7); + CircuitProbe::handled(circuit, 8); + CircuitProbe::handled(circuit, 9); + + // A second on, the sender has had 7 acked and pings with 8 its oldest + // unacked; a resend of 7 that crossed the ack, or a copy the network + // made, may still be on its way. + sMessageTime += F64Seconds(1.0); + circuit.clearDuplicateList(8); + CHECK(CircuitProbe::checkDuplicate(circuit, 7, false)); + CHECK(CircuitProbe::checkDuplicate(circuit, 7, true)); + + // A ping five seconds after those copies lets 7 go, and keeps 8, which + // it does not pass. + sMessageTime += F64Seconds(5.0); + circuit.clearDuplicateList(8); + CHECK_FALSE(CircuitProbe::checkDuplicate(circuit, 7, true)); + CHECK(CircuitProbe::checkDuplicate(circuit, 8, true)); + } + + TEST_CASE("llcircuit: each copy of a handled packet restarts the time a ping keeps its number", "[llcircuit]") + { + StubMessageSystemScope message_system; + LLCircuitData circuit(LLHost(), 0, HEARTBEAT, TIMEOUT); + const F64Seconds handled_at = sMessageTime; + CircuitProbe::handled(circuit, 7); + CircuitProbe::handled(circuit, 8); + CircuitProbe::handled(circuit, 9); + + // The ack for 7 is lost, and the sender resends it. + sMessageTime = handled_at + F64Seconds(4.0); + CHECK(CircuitProbe::checkDuplicate(circuit, 7, true)); + + // The ack for that resend reaches the sender, which pings with 8 its + // oldest unacked, while a resend of 7 it sent before the ack came is + // still on its way. + sMessageTime = handled_at + F64Seconds(6.5); + circuit.clearDuplicateList(8); + sMessageTime = handled_at + F64Seconds(7.0); + CHECK(CircuitProbe::checkDuplicate(circuit, 7, true)); + + // A ping five seconds after that last copy lets 7 go. + sMessageTime = handled_at + F64Seconds(12.0); + circuit.clearDuplicateList(8); + CHECK_FALSE(CircuitProbe::checkDuplicate(circuit, 7, true)); + } +} diff --git a/indra/llmessage/tests/llzerocode_test.cpp b/indra/llmessage/tests/llzerocode_test.cpp index 4ec13cf8f5..d3bc749a30 100644 --- a/indra/llmessage/tests/llzerocode_test.cpp +++ b/indra/llmessage/tests/llzerocode_test.cpp @@ -28,6 +28,8 @@ #include "../llzerocode.h" +#include "message.h" + #include #include @@ -143,7 +145,7 @@ namespace } // decode() must also accept the "0x00 0x00 N" doubled-marker wire format - // (worth +255 zero bytes per extra marker) for compatibility with any + // (worth +256 zero bytes per extra marker) for compatibility with any // encoder other than this one's chunking strategy, even though encode() // itself never emits it. TEST_CASE("LLZeroCode: decode accepts the doubled-marker wrap form", "[LLZeroCode]") @@ -225,6 +227,56 @@ namespace CHECK(dec_size == 0u); } + // Decodes `encoded`, whose header is 6 bytes, into a destination of + // `capacity` bytes, and says how many it wrote: 0 where it overflowed. + U32 decoded_size_within(const std::vector& encoded, U32 capacity, bool& overflow) + { + std::vector dec(capacity, 0xDD); + return LLZeroCode::decode(encoded.data(), (U32)encoded.size(), dec.data(), capacity, 6, overflow); + } + + // A run, as plain bytes may, can end exactly at the destination's + // capacity, and not a byte past it. + TEST_CASE("LLZeroCode: decode fits a final run that ends at the capacity", "[LLZeroCode]") + { + bool overflow = false; + { + INFO("4 plain bytes and a run of 6 after the header: 16 bytes"); + const std::vector encoded = makeBuffer(6, LLZeroCode::FLAG, {1, 2, 3, 4, 0x00, 6}); + CHECK(decoded_size_within(encoded, 16, overflow) == 16u); + CHECK_FALSE(overflow); + CHECK(decoded_size_within(encoded, 15, overflow) == 0u); + CHECK(overflow); + } + { + INFO("a run of 256 + 4 after the header: 266 bytes"); + const std::vector encoded = makeBuffer(6, LLZeroCode::FLAG, {0x00, 0x00, 4}); + CHECK(decoded_size_within(encoded, 266, overflow) == 266u); + CHECK_FALSE(overflow); + CHECK(decoded_size_within(encoded, 265, overflow) == 0u); + CHECK(overflow); + } + } + + // The message system decodes into a buffer of MAX_BUFFER_SIZE bytes, and + // a packet that fills it and ends in zeros decodes whole. + TEST_CASE("LLZeroCode: a packet that fills the receive buffer and ends in zeros round-trips", "[LLZeroCode]") + { + std::vector body(MAX_BUFFER_SIZE - LL_PACKET_ID_SIZE, 0); + body[0] = 1; + const std::vector src = makeBuffer(LL_PACKET_ID_SIZE, 0x00, body); + std::vector enc(2 * src.size(), 0); + const S32 enc_size = LLZeroCode::encode(src.data(), (U32)src.size(), enc.data(), (U32)enc.size(), LL_PACKET_ID_SIZE); + REQUIRE(enc_size > 0); + + std::vector dec(MAX_BUFFER_SIZE, 0xDD); + bool overflow = false; + const U32 dec_size = LLZeroCode::decode(enc.data(), (U32)enc_size, dec.data(), (U32)dec.size(), LL_PACKET_ID_SIZE, overflow); + CHECK_FALSE(overflow); + REQUIRE(dec_size == (U32)MAX_BUFFER_SIZE); + CHECK(memcmp(dec.data(), src.data(), src.size()) == 0); + } + // A pathological alternating zero/non-zero pattern grows under zero-coding // (every isolated zero costs 2 output bytes for 1 input byte), so encode // must refuse it and leave the original buffer in use. diff --git a/indra/llmessage/tests/message_test.cpp b/indra/llmessage/tests/message_test.cpp index c88e29fba1..ac0f2dca20 100644 --- a/indra/llmessage/tests/message_test.cpp +++ b/indra/llmessage/tests/message_test.cpp @@ -43,6 +43,25 @@ #include #endif +namespace ll_test +{ + // What the message system does with a circuit's acks as packets come in + // from its host. + struct CircuitProbe + { + // A reliable packet came in, to be acked. + static void receiveReliable(LLCircuitData& circuit, TPACKETID id) + { + circuit.collectRAck(id); + } + + static size_t acksWaiting(const LLCircuitData& circuit) + { + return circuit.mAcks.size(); + } + }; +} + namespace { struct Response : public LLHTTPNode::Response @@ -155,6 +174,158 @@ namespace CHECK(socket == -1); } + // A message template file of the test's own, there while this lives. + class TemplateFile + { + public: + TemplateFile(const std::string& path, const char* text) + : mPath(path) + { + llofstream file(mPath.c_str()); + file << text; + } + + ~TemplateFile() + { + if (LLFile::remove(mPath) != 0) + { + LL_WARNS() << "Could not remove " << mPath << LL_ENDL; + } + } + + TemplateFile(const TemplateFile&) = delete; + TemplateFile& operator=(const TemplateFile&) = delete; + + private: + std::string mPath; + }; + + // One message, AckBudgetTest, whose one block holds bytes of any length: + // its packet is the 6-byte header, the 4-byte number of a Low message, + // the 2-byte length and the bytes. + const char* const ACK_BUDGET_TEMPLATE = + "version 2.0\n" + "{\n" + " AckBudgetTest Low 1 NotTrusted Unencoded\n" + " {\n" + " Payload Single\n" + " { Data Variable 2 }\n" + " }\n" + "}\n"; + + // gMessageSystem is `system` while this lives, and what it was before + // once it ends. + class UseMessageSystem + { + public: + explicit UseMessageSystem(LLMessageSystem& system) + : mPrior(gMessageSystem) + { + gMessageSystem = &system; + } + + ~UseMessageSystem() + { + gMessageSystem = mPrior; + } + + UseMessageSystem(const UseMessageSystem&) = delete; + UseMessageSystem& operator=(const UseMessageSystem&) = delete; + + private: + LLMessageSystem* mPrior; + }; + + TEST_CASE_METHOD(LLMessageSystemTestData, "LLMessageSystem: acks appended to a packet fit in the MTU with their count", "[LLMessageSystem]") + { + const std::string template_path = mTestConfigDir + mSep + "message_template.msg"; + const TemplateFile message_template(template_path, ACK_BUDGET_TEMPLATE); + LLMessageSystem system(template_path, NET_USE_OS_ASSIGNED_PORT, 1, 0, 0, false, 5.f, 100.f); + REQUIRE(system.isOK()); + const UseMessageSystem use(system); + + // Packets go to the system's own socket, which reads none of them. + const LLHost host("127.0.0.1", system.getListenPort()); + system.enableCircuit(host, false); + LLCircuitData* circuit = system.mCircuitInfo.findCircuit(host); + REQUIRE(circuit != nullptr); + + // Sends AckBudgetTest holding `data_size` bytes, and says how long + // its packet was. + auto send = [&](S32 data_size) + { + const std::vector data(data_size, 0x5a); + system.newMessage("AckBudgetTest"); + system.nextBlock("Payload"); + system.addBinaryData("Data", data.data(), data_size); + return system.sendMessage(host); + }; + + UNSCOPED_INFO("the message alone"); + CHECK(send(1184) == 1196); + + for (TPACKETID id = 1; id <= 30; ++id) + { + ll_test::CircuitProbe::receiveReliable(*circuit, id); + } + + { + INFO("4 bytes to the MTU hold an ack but not the count after it"); + CHECK(send(1184) == 1196); + CHECK(ll_test::CircuitProbe::acksWaiting(*circuit) == 30); + } + { + INFO("5 bytes hold one and its count"); + CHECK(send(1183) == MTUBYTES); + CHECK(ll_test::CircuitProbe::acksWaiting(*circuit) == 29); + } + { + INFO("100 bytes hold 24 and their count"); + CHECK(send(1088) == 1100 + 24 * 4 + 1); + CHECK(ll_test::CircuitProbe::acksWaiting(*circuit) == 5); + } + } + + // The viewer opens every circuit to a simulator with no circuit code, as + // EnableSimulator does, and CloseCircuit, DisableSimulator and a region's + // retirement disable it. + TEST_CASE_METHOD(LLMessageSystemTestData, "LLMessageSystem: disableCircuit removes a circuit opened without a circuit code", "[LLMessageSystem]") + { + const LLHost host("127.0.0.1", 13000); + gMessageSystem->enableCircuit(host, true); + REQUIRE(gMessageSystem->mCircuitInfo.findCircuit(host) != nullptr); + REQUIRE(gMessageSystem->findCircuitCode(host) == 0); + + gMessageSystem->disableCircuit(host); + CHECK(gMessageSystem->mCircuitInfo.findCircuit(host) == nullptr); + + // A later EnableSimulator opens it again. + gMessageSystem->enableCircuit(host, true); + const LLCircuitData* reopened = gMessageSystem->mCircuitInfo.findCircuit(host); + REQUIRE(reopened != nullptr); + CHECK(reopened->isAlive()); + CHECK(reopened->getTrusted()); + } + + TEST_CASE_METHOD(LLMessageSystemTestData, "LLMessageSystem: disableCircuit removes a circuit and the circuit code it has", "[LLMessageSystem]") + { + const LLHost host("127.0.0.1", 13001); + const U32 code = 1234; + const U64 ip_port = ((U64)host.getAddress() << 32) | host.getPort(); + LLUUID session_id; + session_id.generate(); + REQUIRE(gMessageSystem->addCircuitCode(code, session_id)); + gMessageSystem->mIPPortToCircuitCode[ip_port] = code; + gMessageSystem->mCircuitCodeToIPPort[code] = ip_port; + gMessageSystem->enableCircuit(host, false); + REQUIRE(gMessageSystem->findCircuitCode(host) == code); + + gMessageSystem->disableCircuit(host); + CHECK(gMessageSystem->mCircuitInfo.findCircuit(host) == nullptr); + CHECK(gMessageSystem->findCircuitCode(host) == 0); + CHECK(gMessageSystem->mCircuitCodeToIPPort.count(code) == 0); + } + #if !LL_WINDOWS TEST_CASE_METHOD(LLMessageSystemTestData, "LLMessageSystem: start_net closes its socket when it finds no port to bind", "[LLMessageSystem]") // start_net closes the socket it made when it finds no port to bind @@ -201,5 +372,112 @@ namespace CHECK(after == before); } #endif + + // One message, DuplicateTest, whose one block holds a U32. + const char* const DUPLICATE_TEST_TEMPLATE = + "version 2.0\n" + "{\n" + " DuplicateTest Low 1 NotTrusted Unencoded\n" + " {\n" + " Payload Single\n" + " { Data U32 }\n" + " }\n" + "}\n"; + + // A socket of the test's own, closed when this ends. + class SenderSocket + { + public: + SenderSocket() + { + mPort = NET_USE_OS_ASSIGNED_PORT; + mResult = start_net(mSocket, mPort); + } + + ~SenderSocket() + { + end_net(mSocket); + } + + SenderSocket(const SenderSocket&) = delete; + SenderSocket& operator=(const SenderSocket&) = delete; + + S32 mSocket = -1; + int mPort = 0; + S32 mResult = -1; + }; + + S32 sDuplicateTestsHandled = 0; + + void handle_duplicate_test(LLMessageSystem*, void**) + { + ++sDuplicateTestsHandled; + } + + // Reads what has come in until the system has taken `packets` packets + // in all, or a second has passed. + void read_until(LLMessageSystem& system, U32 packets) + { + LockMessageChecker lmc(&system); + const F64 give_up = LLTimer::getTotalSeconds() + 1.0; + while (system.mPacketsIn < packets && LLTimer::getTotalSeconds() < give_up) + { + if (!lmc.checkMessages()) + { + ms_sleep(1); + } + } + } + + TEST_CASE_METHOD(LLMessageSystemTestData, "LLMessageSystem: a reliable packet that arrives twice unmarked is handled once", "[LLMessageSystem]") + { + const std::string template_path = mTestConfigDir + mSep + "message_template.msg"; + const TemplateFile message_template(template_path, DUPLICATE_TEST_TEMPLATE); + LLMessageSystem system(template_path, NET_USE_OS_ASSIGNED_PORT, 1, 0, 0, false, 5.f, 100.f); + REQUIRE(system.isOK()); + const UseMessageSystem use(system); + system.setHandlerFunc("DuplicateTest", handle_duplicate_test); + sDuplicateTestsHandled = 0; + + // The packets come from a socket of the test's own, on a circuit the + // system has to it, as it has to a simulator. + const SenderSocket sender; + REQUIRE(sender.mResult == 0); + system.enableCircuit(LLHost("127.0.0.1", sender.mPort), true); + const LLHost receiver("127.0.0.1", system.getListenPort()); + + // DuplicateTest numbered `id`: the flags, the packet number, no extra + // header, the number of Low message 1 and its U32. + auto send = [&](U8 flags, U8 id) + { + const U8 packet[] = { flags, 0, 0, 0, id, 0, 0xff, 0xff, 0x00, 0x01, 1, 2, 3, 4 }; + REQUIRE(send_packet(sender.mSocket, (const char*)packet, (int)sizeof(packet), receiver.getAddress(), receiver.getPort())); + }; + + { + INFO("both copies in before either is read"); + send(LL_RELIABLE_FLAG, 1); + send(LL_RELIABLE_FLAG, 1); + read_until(system, 2); + REQUIRE(system.mPacketsIn == 2); + CHECK(sDuplicateTestsHandled == 1); + } + { + INFO("the second copy in after the first is handled"); + send(LL_RELIABLE_FLAG, 2); + read_until(system, 3); + send(LL_RELIABLE_FLAG, 2); + read_until(system, 4); + REQUIRE(system.mPacketsIn == 4); + CHECK(sDuplicateTestsHandled == 2); + } + { + INFO("a copy marked resent, as before"); + send(LL_RELIABLE_FLAG | LL_RESENT_FLAG, 2); + read_until(system, 5); + REQUIRE(system.mPacketsIn == 5); + CHECK(sDuplicateTestsHandled == 2); + } + } }