Cache flush lifetimes and a full inbound packet queue

indra/llmessage/llpacketring.cpp

@@ -29,6 +29,7 @@
 #include "llpacketring.h"
 
 #include "llerror.h"
+#include "llframetimer.h"
 
 constexpr S16 MAX_BUFFER_RING_SIZE     = 8192;
 
@@ -59,35 +60,39 @@ LLPacketRing::~LLPacketRing()
     mHeadIndex = 0;
 }
 
-void LLPacketRing::pushPacket(const LLPacketBuffer& packet)
+bool LLPacketRing::hasRoom() const
 {
-    S16 ring_size = (S16)mRing.size();
-    if (mNumBufferedPackets >= ring_size && ring_size < MAX_BUFFER_RING_SIZE)
+    return mNumBufferedPackets < (S16)mRing.size() || (S16)mRing.size() < MAX_BUFFER_RING_SIZE;
+}
+
+bool LLPacketRing::pushPacket(const LLPacketBuffer& packet)
+{
+    if (mNumBufferedPackets >= (S16)mRing.size() && !expandRing())
     {
-        expandRing();
-        ring_size = (S16)mRing.size();
+        // A flood that fills the ring leaves a packet out with every one
+        // that comes in, so it is told of a second at a time.
+        ++mLeftOutSinceWarning;
+        const F64 now = LLFrameTimer::getTotalSeconds();
+        if (!mWarnedLeftOut || now - mLastLeftOutWarning >= 1.0)
+        {
+            LL_WARNS("PacketRing") << "buffer full at " << mNumBufferedPackets << " packets, " << mLeftOutSinceWarning
+                                   << " left out" << LL_ENDL;
+            mWarnedLeftOut = true;
+            mLastLeftOutWarning = now;
+            mLeftOutSinceWarning = 0;
+        }
+        return false;
     }
+    const S16 ring_size = (S16)mRing.size();
 
     LLPacketBuffer* slot = mRing[mHeadIndex];
-    S32 old_size = slot->getSize();
-
     *slot = packet;
 
     mHeadIndex = (mHeadIndex + 1) % ring_size;
 
-    if (mNumBufferedPackets < ring_size)
-    {
-        ++mNumBufferedPackets;
-        mNumBufferedBytes += packet.getSize();
-    }
-    else
-    {
-        // Ring is at maximum capacity; oldest packet was overwritten.
-        // This is VERY BAD because we've already ACKed the packet we're loosing
-        // (if it was "reliable").
-        LL_WARNS("PacketRing") << "buffer overflow at " << mNumBufferedPackets << " packets" << LL_ENDL;
-        mNumBufferedBytes += packet.getSize() - old_size;
-    }
+    ++mNumBufferedPackets;
+    mNumBufferedBytes += packet.getSize();
+    return true;
 }
 
 bool LLPacketRing::popPacket(LLPacketBuffer& packet)

indra/llmessage/llpacketring.h

@@ -8,9 +8,8 @@
  * popPacket().
  *
  * The ring starts at DEFAULT_BUFFER_RING_SIZE slots and grows in increments
- * of BUFFER_RING_EXPANSION up to MAX_BUFFER_RING_SIZE.  Once at the ceiling,
- * pushPacket() silently overwrites the oldest queued packet to make room for
- * the incoming one.
+ * of BUFFER_RING_EXPANSION up to MAX_BUFFER_RING_SIZE.  Once full at the
+ * ceiling, pushPacket() refuses the incoming packet.
  *
  * $LicenseInfo:firstyear=2001&license=viewerlgpl$
  * Second Life Viewer Source Code
@@ -47,9 +46,14 @@ class LLPacketRing
     LLPacketRing();
     ~LLPacketRing();
 
-    // Copy 'packet' onto the tail of the ring, growing the ring or
-    // overwriting the oldest entry when the ring is at capacity.
-    void pushPacket(const LLPacketBuffer& packet);
+    // Copy 'packet' onto the tail of the ring, growing the ring when it is
+    // full. A ring full at its ceiling leaves the packet out and returns
+    // false: the oldest packet it holds may be a reliable one already acked,
+    // which its sender would never send again.
+    bool pushPacket(const LLPacketBuffer& packet);
+
+    // Whether pushPacket() would take a packet now.
+    bool hasRoom() const;
 
     // Copy the head (oldest) packet into 'packet'.
     // Returns true if a packet was available, false if the ring was empty.
@@ -69,4 +73,8 @@ class LLPacketRing
     S16 mHeadIndex          { 0 };
     S16 mNumBufferedPackets { 0 };
     S32 mNumBufferedBytes   { 0 };
+
+    S32  mLeftOutSinceWarning { 0 };
+    F64  mLastLeftOutWarning  { 0.0 };
+    bool mWarnedLeftOut       { false };
 };

indra/llmessage/message.cpp

@@ -1159,31 +1159,6 @@ S32 LLMessageSystem::bufferInboundPacket()
             }
         }
 
-        if (cdp)
-        {
-            // ACK inbound reliable packet ASAP, so its sender stops resending
-            // it while it waits in the queue, whatever checkMessages() makes
-            // of it, which does not ack it again.
-            if ((data[0] & LL_RELIABLE_FLAG))
-            {
-                cdp->collectRAck(recv_packet_id);
-                pkt.setAcked(true);
-            }
-
-            // Check packet sequencing here, in true socket-arrival order, before
-            // this packet is sorted into the high/low priority inbound queue.
-            // 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;
-            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);
-            }
-        }
-
         auto enqueue = [this](const LLPacketBuffer& buffered)
         {
             if (isHighPriorityMessage(buffered))
@@ -1196,13 +1171,54 @@ S32 LLMessageSystem::bufferInboundPacket()
             }
         };
 
+        auto take = [&](LLPacketBuffer& arrived)
+        {
+            // A queue full at its ceiling has no room for the packet, which is
+            // left out as the network might have dropped it: neither acked nor
+            // its number recorded, so that its sender sends it again.
+            const bool has_room = (isHighPriorityMessage(arrived) ? mHighPriorityInbound : mLowPriorityInbound).hasRoom();
+
+            if (cdp)
+            {
+                // Skip duplicates of reliable packets already handled, marked resent
+                // or not, same as checkMessages()/logValidMsg() would do further
+                // downstream.
+                const bool recv_resent = (data[0] & LL_RESENT_FLAG) != 0;
+                const bool recv_reliable = (data[0] & LL_RELIABLE_FLAG) != 0;
+                const bool duplicate = recv_reliable && cdp->checkDuplicate(recv_packet_id, recv_resent);
+
+                // ACK inbound reliable packet ASAP, so its sender stops resending
+                // it while it waits in the queue, whatever checkMessages() makes
+                // of it, which does not ack it again. A copy of a packet already
+                // taken is acked with no room for it, as it would be with room.
+                if (recv_reliable && (has_room || duplicate))
+                {
+                    cdp->collectRAck(recv_packet_id);
+                    arrived.setAcked(true);
+                }
+
+                // Check packet sequencing here, in true socket-arrival order, before
+                // this packet is sorted into the high/low priority inbound queue.
+                if (has_room && !duplicate)
+                {
+                    cdp->checkPacketInID(recv_packet_id, recv_resent);
+                    arrived.setPacketIDChecked(true);
+                }
+            }
+
+            enqueue(arrived);
+        };
+
 #if AL_NET_IMPAIRMENT
         std::vector<LLPacketBuffer> released;
         mNetImpairment.advance(released);
 
+        // A packet held back arrives when it is released: checkMessages() acks
+        // it and checks its number as it reads it, and a queue with no room for
+        // it then leaves it out unacked.
         if (!mNetImpairment.hold(pkt))
         {
-            enqueue(pkt);
+            take(pkt);
         }
 
         // Anything released rejoins *behind* the packet that just arrived, which is what puts it
@@ -1212,7 +1228,7 @@ S32 LLMessageSystem::bufferInboundPacket()
             enqueue(late);
         }
 #else
-        enqueue(pkt);
+        take(pkt);
 #endif
     }
 

indra/llmessage/tests/message_test.cpp

@@ -36,6 +36,7 @@
 #include "net.h"
 
 #include <algorithm>
+#include <array>
 
 #if !LL_WINDOWS
 #include <fcntl.h>
@@ -67,6 +68,12 @@ namespace ll_test
         {
             return std::count(circuit.mAcks.begin(), circuit.mAcks.end(), id);
         }
+
+        // The number the circuit takes the next packet in order to have.
+        static TPACKETID packetsInID(const LLCircuitData& circuit)
+        {
+            return circuit.mPacketsInID;
+        }
     };
 }
 
@@ -752,5 +759,113 @@ namespace
         }
         CHECK(ll_test::CircuitProbe::acksWaiting(*circuit) == 5);
     }
+
+    // The packets a full queue holds were acked as they came in, so taking
+    // the place of the oldest would lose a packet its sender never sends
+    // again. The packet the queue has no room for is left out unacked.
+    TEST_CASE_METHOD(LLMessageSystemTestData, "LLMessageSystem: a full inbound queue leaves a reliable packet out unacked", "[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;
+
+        const SenderSocket sender;
+        REQUIRE(sender.mResult == 0);
+        const LLHost sender_host("127.0.0.1", sender.mPort);
+        system.enableCircuit(sender_host, true);
+        const LLCircuitData* circuit = system.mCircuitInfo.findCircuit(sender_host);
+        REQUIRE(circuit != nullptr);
+        const LLHost receiver("127.0.0.1", system.getListenPort());
+
+        // A packet numbered `id` carrying message number `message`.
+        using MessageNumber = std::array<U8, 4>;
+        auto send = [&](U32 id, U8 flags, const MessageNumber& message)
+        {
+            const U8 packet[] = { flags, (U8)(id >> 24), (U8)(id >> 16), (U8)(id >> 8), (U8)id, 0,
+                                  message[0], message[1], message[2], message[3], 1, 2, 3, 4 };
+            REQUIRE(send_packet(sender.mSocket, (const char*)packet, (int)sizeof(packet), receiver.getAddress(), receiver.getPort()));
+        };
+
+        // Waits until the system has left `packets` packets out in all.
+        auto drain_until_left_out = [&](S32 packets)
+        {
+            const F64 give_up = LLTimer::getTotalSeconds() + 1.0;
+            while (system.getTotalNumDroppedPackets() < packets && LLTimer::getTotalSeconds() < give_up)
+            {
+                system.drainUdpSocket();
+                ms_sleep(1);
+            }
+            REQUIRE(system.getTotalNumDroppedPackets() == packets);
+        };
+
+        // Fills the ring `message` goes to up to its ceiling with reliable
+        // packets numbered from `first`, a few at a time so that the socket's
+        // buffer holds every one until it is queued, and sends one more,
+        // which is left out. Returns its number.
+        constexpr S32 FULL = 8192;
+        auto overflow = [&](U32 first, const MessageNumber& message) -> U32
+        {
+            const size_t acks_before = ll_test::CircuitProbe::acksWaiting(*circuit);
+            U32 id = first;
+            while (system.getNumBufferedPackets() < FULL)
+            {
+                const S32 batch = llmin(128, FULL - system.getNumBufferedPackets());
+                for (S32 i = 0; i < batch; ++i)
+                {
+                    send(id++, LL_RELIABLE_FLAG, message);
+                }
+                queue_until(system, (S32)(id - first));
+                REQUIRE(system.getNumBufferedPackets() == (S32)(id - first));
+            }
+            REQUIRE(ll_test::CircuitProbe::acksWaiting(*circuit) == acks_before + FULL);
+
+            const U32 left_out = id;
+            send(left_out, LL_RELIABLE_FLAG, message);
+            drain_until_left_out(1);
+            CHECK(system.getNumBufferedPackets() == FULL);
+            CHECK(ll_test::CircuitProbe::acksWaitingFor(*circuit, left_out) == 0);
+            CHECK(ll_test::CircuitProbe::acksWaiting(*circuit) == acks_before + FULL);
+            UNSCOPED_INFO("nor is its number taken as come");
+            CHECK(ll_test::CircuitProbe::packetsInID(*circuit) == left_out);
+            return left_out;
+        };
+
+        SECTION("the low priority ring")
+        {
+            // DuplicateTest, Low 1.
+            const MessageNumber duplicate_test = { 0xff, 0xff, 0x00, 0x01 };
+            send(1, LL_RELIABLE_FLAG, duplicate_test);
+            read_until(system, 1);
+            REQUIRE(sDuplicateTestsHandled == 1);
+            const U32 left_out = overflow(2, duplicate_test);
+
+            UNSCOPED_INFO("a copy of a packet already read is acked all the same, as one it would drop");
+            send(1, LL_RELIABLE_FLAG | LL_RESENT_FLAG, duplicate_test);
+            drain_until_left_out(2);
+            CHECK(ll_test::CircuitProbe::acksWaitingFor(*circuit, 1) == 2);
+
+            UNSCOPED_INFO("every packet queued is read, the first among them");
+            read_queued(system);
+            CHECK(sDuplicateTestsHandled == FULL + 1);
+
+            UNSCOPED_INFO("and the one left out is taken when it comes again");
+            send(left_out, LL_RELIABLE_FLAG | LL_RESENT_FLAG, duplicate_test);
+            read_until(system, FULL + 2);
+            CHECK(sDuplicateTestsHandled == FULL + 2);
+            CHECK(ll_test::CircuitProbe::acksWaitingFor(*circuit, left_out) == 1);
+        }
+
+        SECTION("the high priority ring")
+        {
+            // A Medium message, whose bytes after the number are none that
+            // isHighPriorityMessage keeps in the low priority ring.
+            const MessageNumber medium = { 0xff, 0x01, 0x01, 0x01 };
+            overflow(1, medium);
+        }
+    }
 }
 

indra/newview/llviewerregion.cpp

@@ -2962,6 +2962,14 @@ LLViewerRegion::eCacheUpdateResult LLViewerRegion::cacheFullUpdate(LLDataPackerB
     LLViewerObject::unpackU32(&dp, local_id, "LocalID");
     LLViewerObject::unpackU32(&dp, crc, "CRC");
 
+    // Nor is one sent whole kept in it, to be let go before it is made: it is
+    // asked for again, to come once the cache has been removed.
+    if (cacheRemovalRequested())
+    {
+        addCacheMissFull(local_id);
+        return CACHE_UPDATE_DUPE;
+    }
+
     LLVOCacheEntry* entry = getCacheEntry(local_id, false);
 
     if (entry)
@@ -3100,10 +3108,24 @@ void LLViewerRegion::addToCreatedList(U32 local_id)
 
 // Get data packer for this object, if we have cached data
 // AND the CRC matches. JC
+bool LLViewerRegion::cacheRemovalRequested() const
+{
+    return LLVOCache::instanceExists() && LLVOCache::getInstance()->isRemovalRequested(mHandle);
+}
+
 bool LLViewerRegion::probeCache(U32 local_id, U32 crc, U32 flags, U8 &cache_miss_type)
 {
     //llassert(mCacheLoaded);  This assert failes often, changing to early-out -- davep, 2010/10/18
 
+    // An object a cache about to be removed holds would never be made from it,
+    // so it is asked for as the removed cache would have it asked for.
+    if (cacheRemovalRequested())
+    {
+        addCacheMiss(local_id, CACHE_MISS_TYPE_TOTAL);
+        cache_miss_type = CACHE_MISS_TYPE_TOTAL;
+        return false;
+    }
+
     LLVOCacheEntry* entry = getCacheEntry(local_id, false);
 
     if (entry)
@@ -3256,7 +3278,15 @@ void LLViewerRegion::dumpCache()
 
 void LLViewerRegion::clearVOCacheFromMemory()
 {
-    mImpl->mCacheMap.clear();
+    // The map holds the only references to its entries, and the sets, the
+    // waiting list, the cache's tree, the drawables' octree entries and the
+    // eviction walk's cursor hold bare pointers to them.
+    std::vector<LLPointer<LLVOCacheEntry>> released;
+    LLVOCacheEntry::releaseAll(mImpl->mCacheMap, { &mImpl->mActiveSet, &mImpl->mWaitingSet, &mImpl->mVisibleEntries },
+                               [this](LLVOCacheEntry* entry) { removeFromVOCacheTree(entry); },
+                               LLAppViewer::instance()->isQuitting() ? released : sRegionCacheCleanup);
+    mImpl->mWaitingList.clear();
+    mLastVisitedEntry = NULL;
 }
 
 void LLViewerRegion::unpackRegionHandshake()