The reliable UDP circuit times its resends and acks as meant, and closes and dedupes as meant

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.

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

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;

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

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.

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<TPACKETID> 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);
 }
 
 

indra/llmessage/message.h

@@ -157,10 +157,11 @@ class LLMessageStringTable : public LLSingleton<LLMessageStringTable>
 //
 // 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.