Fix PABotBase2 retransmits.

Author: Mysticial

Common/Cpp/StreamConnections/MockDevice.cpp

@@ -5,6 +5,7 @@
  */
 
 #include <string.h>
+#include "Common/Cpp/Exceptions.h"
 #include "Common/PABotBase2/PABotBase2_ConnectionDebug.h"
 #include "MockDevice.h"
 
@@ -13,7 +14,7 @@
 using std::cout;
 using std::endl;
 
-#if 0
+#if 1
 #define PABB2_DROP_HOST_TO_DEVICE   0.2
 #define PABB2_DROP_DEVICE_TO_HOST   0.2
 #else
@@ -158,10 +159,17 @@ size_t MockDevice::verify_stream_data(){
         actual.data(), bytes
     );
 
+    if (read == 0){
+        return bytes;
+    }
+
+
+#if 0
     {
         std::lock_guard<Mutex> lg1(m_print_lock);
         cout << "read = " << read << endl;
     }
+#endif
 
 
     std::vector<uint8_t> expected(
@@ -178,17 +186,20 @@ size_t MockDevice::verify_stream_data(){
 
     std::lock_guard<Mutex> lg1(m_print_lock);
     if (matched){
+#if 0
         cout << "Matched: Bytes = " << read
              << ", Remaining = " << m_expected_host_to_device_stream.size()
              << ", Matched = "
              << std::string((const char*)actual.data(), read)
              << endl;
+#endif
     }else{
         cout << "MISMATCH: Expected = "
              << std::string((const char*)expected.data(), read)
              << ", Actual = "
              << std::string((const char*)actual.data(), read)
              << endl;
+        throw InternalProgramError(nullptr, PA_CURRENT_FUNCTION, "Mismatch");
     }
     return m_expected_host_to_device_stream.size();
 }
@@ -200,7 +211,7 @@ void MockDevice::device_thread(){
     std::unique_lock<Mutex> lg(m_device_lock);
     while (!m_stopping.load(std::memory_order_relaxed)){
         pabb2_ReliableStreamConnection_run_events(&m_connection);
-        m_device_cv.wait(lg);
+        m_device_cv.wait_for(lg, Milliseconds(10));
     }
 }
 void MockDevice::host_recv_thread(){

Common/Cpp/StreamConnections/ReliableStreamConnection.cpp

@@ -11,10 +11,9 @@
 #include "Common/Cpp/StreamConnections/PABotBase2_MessageDumper.h"
 #include "ReliableStreamConnection.h"
 
-//  REMOVE
-#include <iostream>
-using std::cout;
-using std::endl;
+//#include <iostream>
+//using std::cout;
+//using std::endl;
 
 namespace PokemonAutomation{
 
@@ -234,10 +233,12 @@ void ReliableStreamConnection::retransmit_thread(){
 //
 
 void ReliableStreamConnection::on_recv(const void* data, size_t bytes){
+#if 0
     if (m_print_lock){
         std::lock_guard<Mutex> lg(*m_print_lock);
         cout << "ReliableStreamConnection::on_recv(): " << bytes << endl;
     }
+#endif
     pabb2_PacketParser_push_bytes(
         &m_parser,
         this, &ReliableStreamConnection::on_packet,

Common/PABotBase2/PABotBase2_PacketSender.c

@@ -286,29 +286,42 @@ bool pabb2_PacketSender_iterate_retransmits(pabb2_PacketSender* self){
         size_t offset = ~self->offsets[head & PABB2_ConnectionSender_SLOTS_MASK];
         pabb2_PacketHeader* packet = (pabb2_PacketHeader*)(self->buffer + offset);
 
-        //  Retransmit if it hasn't been acked already and is old enough.
-        if (packet->opcode != PABB2_CONNECTION_OPCODE_INVALID &&
-            seqnum - packet->magic_number >= PABB2_ConnectionSender_RETRANSMIT_COUNTER
-        ){
-            packet->magic_number = PABB2_CONNECTION_MAGIC_NUMBER;
-            uint8_t packet_bytes = packet->packet_bytes;
-
+        //  Already acked.
+        if (packet->opcode == PABB2_CONNECTION_OPCODE_INVALID){
 #if 0
-            printf("Retransmitting: %u\n", packet->seqnum);
+            printf("Already Acked\n");
             fflush(stdout);
 #endif
+            head++;
+            continue;
+        }
 
-            self->unreliable_sender_send(
-                self->unreliable_sender_context,
-                packet,
-                packet_bytes == 0 ? (size_t)256 : (size_t)packet_bytes,
-                true
-            );
-            packet->magic_number = seqnum;
-            return true;
+        //  Not old enough.
+        if ((uint8_t)(seqnum - packet->magic_number) < PABB2_ConnectionSender_RETRANSMIT_COUNTER){
+#if 0
+            printf("Not old enough.\n");
+            fflush(stdout);
+#endif
+            head++;
+            continue;
         }
 
-        head++;
+        packet->magic_number = PABB2_CONNECTION_MAGIC_NUMBER;
+        uint8_t packet_bytes = packet->packet_bytes;
+
+#if 0
+        printf("Retransmitting: %u\n", packet->seqnum);
+        fflush(stdout);
+#endif
+
+        self->unreliable_sender_send(
+            self->unreliable_sender_context,
+            packet,
+            packet_bytes == 0 ? (size_t)256 : (size_t)packet_bytes,
+            true
+        );
+        packet->magic_number = seqnum;
+        return true;
     }
 
     //  Increment counter only if we did nothing.

SerialPrograms/Source/NintendoSwitch/DevPrograms/TestProgramComputer.cpp

@@ -316,17 +316,29 @@ void stress_test(Logger& logger, CancellableScope& scope){
     connection.send_request(PABB2_CONNECTION_OPCODE_ASK_BUFFER_SLOTS);
     connection.wait_for_pending();
 
+    uint64_t bytes_sent = 0;
+    WallClock last_print = current_time();
+
     while (true){
         scope.throw_if_cancelled();
 
         std::string data = random_string(20);
         const char* ptr = data.data();
         size_t left = data.size();
         while (left > 0){
+//            scope.wait_for(Milliseconds(rand() % 100));
             size_t sent = connection.send(ptr, left);
+            if (sent == 0){
+                device.verify_stream_data();
+            }
             ptr += sent;
             left -= sent;
         }
+        bytes_sent += data.size();
+        if (current_time() - last_print > Seconds(1)){
+            cout << "Bytes Sent = " << bytes_sent << endl;
+            last_print = current_time();
+        }
         device.push_expected_stream_data(data.data(), data.size());
     }
 }