More work on PABotBase2.

Author: Mysticial

Common/Cpp/Concurrency/Backends/Thread_Qt.tpp

@@ -4,13 +4,13 @@
  *
  */
 
+#include <iostream>
 #include <QCoreApplication>
 #include <QThread>
 #include "Common/Cpp/Containers/Pimpl.tpp"
 #include "Common/Cpp/Concurrency/Qt6.9ThreadBugWorkaround.h"
 #include "Common/Cpp/Concurrency/Thread.h"
 
-//#include <iostream>
 //using std::cout;
 //using std::endl;
 
@@ -110,6 +110,7 @@ void Thread::join(){
     m_data->quit();
 
     //  But instead of waiting and joining (which will hang), we leak it intentionally.
+    std::cout << "Intentionally leaking QThread to work around Qt6.9 thread adoption bug..." << std::endl;
     m_data.release();
 #endif
 

Common/Cpp/StreamConnections/MockDevice.cpp

@@ -52,6 +52,7 @@ MockDevice::~MockDevice(){
 }
 void MockDevice::print() const{
     std::lock_guard<Mutex> lg(m_device_lock);
+    std::lock_guard<Mutex> lg1(m_print_lock);
     pabb2_StreamCoalescer_print(&m_connection.stream_coalescer, true);
 }
 
@@ -65,6 +66,7 @@ size_t MockDevice::device_send_serial(const void* data, size_t bytes, bool is_re
         }
 
         if ((rand() % 100) / 100. < PABB2_DROP_DEVICE_TO_HOST){
+            std::lock_guard<Mutex> lg1(m_print_lock);
             cout << "**Intentionally Dropping Packet: device -> host**" << endl;
             return 0;
         }
@@ -77,6 +79,7 @@ size_t MockDevice::device_send_serial(const void* data, size_t bytes, bool is_re
         );
 
         if ((rand() % 100) / 100. < PABB2_DROP_DEVICE_TO_HOST){
+            std::lock_guard<Mutex> lg1(m_print_lock);
             cout << "**Intentionally Corrupting Packet: device -> host**" << endl;
             m_device_to_host_line[rand() % m_device_to_host_line.size()] = 0;
         }
@@ -104,6 +107,7 @@ size_t MockDevice::send(const void* data, size_t bytes){
 //        cout << "MockDevice::send(const void* data, size_t bytes)" << endl;
 
         if ((rand() % 100) / 100. < PABB2_DROP_HOST_TO_DEVICE){
+            std::lock_guard<Mutex> lg1(m_print_lock);
             cout << "**Intentionally Dropping Packet: host -> device**" << endl;
             return 0;
         }
@@ -119,6 +123,7 @@ size_t MockDevice::send(const void* data, size_t bytes){
         );
 
         if ((rand() % 100) / 100. < PABB2_DROP_HOST_TO_DEVICE){
+            std::lock_guard<Mutex> lg1(m_print_lock);
             cout << "**Intentionally Corrupting Packet: host -> device**" << endl;
             m_host_to_device_line[rand() % m_host_to_device_line.size()] = 0;
         }
@@ -131,6 +136,53 @@ size_t MockDevice::send(const void* data, size_t bytes){
     return bytes;
 }
 
+void MockDevice::push_expected_stream_data(const void* data, size_t bytes){
+    std::lock_guard<Mutex> lg(m_device_lock);
+
+    m_expected_host_to_device_stream.insert(
+        m_expected_host_to_device_stream.end(),
+        (const uint8_t*)data,
+        (const uint8_t*)data + bytes
+    );
+
+    bytes = m_expected_host_to_device_stream.size();
+
+    std::vector<uint8_t> actual(bytes);
+    size_t read = pabb2_ReliableStreamConnection_read_stream(
+        &m_connection,
+        actual.data(), bytes
+    );
+
+    {
+        std::lock_guard<Mutex> lg1(m_print_lock);
+        cout << "read = " << read << endl;
+    }
+
+
+    std::vector<uint8_t> expected(
+        m_expected_host_to_device_stream.begin(),
+        m_expected_host_to_device_stream.begin() + read
+    );
+
+    m_expected_host_to_device_stream.erase(
+        m_expected_host_to_device_stream.begin(),
+        m_expected_host_to_device_stream.begin() + read
+    );
+
+    if (memcmp(actual.data(), expected.data(), read) == 0){
+        return;
+    }
+
+    std::lock_guard<Mutex> lg1(m_print_lock);
+    cout << "MISMATCH: Expected: "
+         << std::string((const char*)expected.data(), read)
+         << ", Actual: "
+         << std::string((const char*)actual.data(), read)
+         << endl;
+}
+
+
+
 
 void MockDevice::device_thread(){
     std::unique_lock<Mutex> lg(m_device_lock);

Common/Cpp/StreamConnections/MockDevice.h

@@ -27,6 +27,10 @@ class MockDevice : public StreamConnection{
 
     void print() const;
 
+    Mutex& print_lock() const{
+        return m_print_lock;
+    }
+
 
 private:
     //  Call from device.
@@ -54,6 +58,8 @@ class MockDevice : public StreamConnection{
 
     virtual size_t send(const void* data, size_t bytes) override;
 
+    void push_expected_stream_data(const void* data, size_t bytes);
+
 
 private:
     void device_thread();
@@ -71,6 +77,8 @@ class MockDevice : public StreamConnection{
     size_t m_host_to_device_capacity = 1024;
     std::deque<uint8_t> m_host_to_device_line;
 
+    std::deque<uint8_t> m_expected_host_to_device_stream;
+
     std::atomic<bool> m_stopping;
 
     mutable Mutex m_device_lock;
@@ -80,6 +88,8 @@ class MockDevice : public StreamConnection{
     mutable Mutex m_host_lock;
     ConditionVariable m_host_cv;
     AsyncTask m_host_thread;
+
+    mutable Mutex m_print_lock;
 };
 
 

Common/Cpp/StreamConnections/ReliableStreamConnection.cpp

@@ -25,11 +25,13 @@ ReliableStreamConnection::ReliableStreamConnection(
     Logger& logger, bool log_everything,
     ThreadPool& thread_pool,
     StreamConnection& unreliable_connection,
-    WallDuration retransmit_timeout
+    WallDuration retransmit_timeout,
+    Mutex* print_lock
 )
     : m_logger(logger)
     , m_unreliable_connection(unreliable_connection)
     , m_retransmit_timeout(retransmit_timeout)
+    , m_print_lock(print_lock)
     , m_log_everything(log_everything)
 //    , m_version_verified(false)
     , m_remote_slot_capacity(1)
@@ -228,7 +230,10 @@ void ReliableStreamConnection::retransmit_thread(){
 //
 
 void ReliableStreamConnection::on_recv(const void* data, size_t bytes){
-    cout << "ReliableStreamConnection::on_recv(): " << bytes << endl;
+    if (m_print_lock){
+        std::lock_guard<Mutex> lg(*m_print_lock);
+        cout << "ReliableStreamConnection::on_recv(): " << bytes << endl;
+    }
     pabb2_PacketParser_push_bytes(
         &m_parser,
         this, &ReliableStreamConnection::on_packet,

Common/Cpp/StreamConnections/ReliableStreamConnection.h

@@ -34,7 +34,8 @@ class ReliableStreamConnection final
         Logger& logger, bool log_everything,
         ThreadPool& thread_pool,
         StreamConnection& unreliable_connection,
-        WallDuration retransmit_timeout = Milliseconds(100)
+        WallDuration retransmit_timeout = Milliseconds(100),
+        Mutex* print_lock = nullptr
     );
     ~ReliableStreamConnection();
 
@@ -98,6 +99,7 @@ class ReliableStreamConnection final
     Logger& m_logger;
     StreamConnection& m_unreliable_connection;
     const WallDuration m_retransmit_timeout;
+    Mutex* m_print_lock;
 
     pabb2_PacketSender m_reliable_sender;
     pabb2_PacketParser m_parser;
@@ -113,6 +115,7 @@ class ReliableStreamConnection final
     mutable Mutex m_lock;
     ConditionVariable m_cv;
     AsyncTask m_retransmit_thread;
+
 };
 
 

Common/PABotBase2/PABotBase2_StreamCoalescer.c

@@ -144,7 +144,7 @@ bool pabb2_StreamCoalescer_push_stream(pabb2_StreamCoalescer* self, const pabb2_
         uint8_t diff = seqnum - slot_head;
 //        printf("seqnum = %d, slot_head = %d\n", seqnum, slot_head);
         if (diff >= PABB2_StreamCoalescer_SLOTS){
-//            printf("Device: In the past.\n");
+//            printf("Device: In the past or too far ahead.\n");
             //  Negative means we're in the past and we can just ack.
             return diff & 0x80;
         }
@@ -185,7 +185,7 @@ bool pabb2_StreamCoalescer_push_stream(pabb2_StreamCoalescer* self, const pabb2_
 size_t pabb2_StreamCoalescer_read(pabb2_StreamCoalescer* self, void* data, size_t max_bytes){
     if (self->stream_reset){
         self->stream_reset = false;
-        return (size_t)-1;
+        return 0;
     }
 
     size_t read = 0;

SerialPrograms/Source/NintendoSwitch/DevPrograms/TestProgramComputer.cpp

@@ -309,7 +309,8 @@ void TestProgramComputer::program(ProgramEnvironment& env, CancellableScope& sco
             logger, true,
             GlobalThreadPools::unlimited_realtime(),
             device,
-            1s
+            1s,
+            &device.print_lock()
         );
 
         connection.reset();
@@ -324,19 +325,21 @@ void TestProgramComputer::program(ProgramEnvironment& env, CancellableScope& sco
         connection.wait_for_pending();
 
 
-        connection.send("asdf", 4);
-        connection.send("qwer", 4);
-        connection.send("zxcv", 4);
+        connection.send("asdf", 4); device.push_expected_stream_data("asdf", 4);
+        connection.send("qwer", 4); device.push_expected_stream_data("qwer", 4);
+        connection.send("zxcv", 4); device.push_expected_stream_data("zxcv", 4);
+        connection.wait_for_pending();
+
         cout << "sent = " << connection.send("0123456789abcdef", 16) << endl;
 //        connection.send("0123456789abcdef", 16);
 //        connection.send("0123456789abcdef", 16);
 //        connection.send("0123456789abcdef", 16);
 
         connection.print();
+        device.print();
 
         connection.wait_for_pending();
 
-        device.print();
 
         scope.wait_for(60s);
     }

SerialPrograms/Source/PokemonLZA/Programs/PokemonLZA_FastTravelNavigation.cpp

@@ -368,7 +368,11 @@ bool set_fast_travel_menu_filter_routine(
     int target_filter_index = static_cast<int>(filter);
     WallClock deadline = current_time() + 30s;
     do {
-        int selected_filter_index = get_current_selector_index(console, FAST_TRAVEL_FILTER_ARROW_BOX(), FAST_TRAVEL_FILTER_SPACING);
+        int selected_filter_index = get_current_selector_index(
+            console,
+            FAST_TRAVEL_FILTER_ARROW_BOX(),
+            FAST_TRAVEL_FILTER_SPACING
+        );
         if (selected_filter_index == -1){
             return false;
         }
@@ -557,4 +561,4 @@ FastTravelState open_map_and_fly_to(ConsoleHandle& console, ProControllerContext
 
 }
 }
-}
+}