More work on PABotBase2.

Author: Mysticial

Common/Cpp/StreamConnections/MockDevice.cpp

@@ -13,7 +13,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
@@ -137,15 +137,20 @@ size_t MockDevice::send(const void* data, size_t 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
-    );
+    {
+        std::lock_guard<Mutex> lg(m_device_lock);
 
-    bytes = m_expected_host_to_device_stream.size();
+        m_expected_host_to_device_stream.insert(
+            m_expected_host_to_device_stream.end(),
+            (const uint8_t*)data,
+            (const uint8_t*)data + bytes
+        );
+    }
+    verify_stream_data();
+}
+size_t MockDevice::verify_stream_data(){
+    std::lock_guard<Mutex> lg(m_device_lock);
+    size_t bytes = m_expected_host_to_device_stream.size();
 
     std::vector<uint8_t> actual(bytes);
     size_t read = pabb2_ReliableStreamConnection_read_stream(
@@ -169,16 +174,23 @@ void MockDevice::push_expected_stream_data(const void* data, size_t bytes){
         m_expected_host_to_device_stream.begin() + read
     );
 
-    if (memcmp(actual.data(), expected.data(), read) == 0){
-        return;
-    }
+    bool matched = memcmp(actual.data(), expected.data(), read) == 0;
 
     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;
+    if (matched){
+        cout << "Matched: Bytes = " << read
+             << ", Remaining = " << m_expected_host_to_device_stream.size()
+             << ", Matched = "
+             << std::string((const char*)actual.data(), read)
+             << endl;
+    }else{
+        cout << "MISMATCH: Expected = "
+             << std::string((const char*)expected.data(), read)
+             << ", Actual = "
+             << std::string((const char*)actual.data(), read)
+             << endl;
+    }
+    return m_expected_host_to_device_stream.size();
 }
 
 

Common/Cpp/StreamConnections/MockDevice.h

@@ -59,6 +59,7 @@ 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);
+    size_t verify_stream_data();
 
 
 private:

Common/Cpp/StreamConnections/ReliableStreamConnection.cpp

@@ -74,6 +74,10 @@ bool ReliableStreamConnection::cancel(std::exception_ptr exception) noexcept{
     m_cv.notify_all();
     return false;
 }
+size_t ReliableStreamConnection::pending() const{
+    std::unique_lock<Mutex> lg(m_lock);
+    return pabb2_PacketSender_slots_used(&m_reliable_sender);
+}
 void ReliableStreamConnection::wait_for_pending(){
     std::unique_lock<Mutex> lg(m_lock);
     m_cv.wait(lg, [this]{

Common/Cpp/StreamConnections/ReliableStreamConnection.h

@@ -43,6 +43,7 @@ class ReliableStreamConnection final
         cancel(nullptr);
     }
     virtual bool cancel(std::exception_ptr exception) noexcept override;
+    size_t pending() const;
     void wait_for_pending();
 
     void reset();

Common/PABotBase2/PABotBase2_PacketSender.h

@@ -75,10 +75,10 @@ void pabb2_PacketSender_init(
 );
 void pabb2_PacketSender_reset(pabb2_PacketSender* self);
 
-static inline uint8_t pabb2_PacketSender_slots_used(pabb2_PacketSender* self){
+static inline uint8_t pabb2_PacketSender_slots_used(const pabb2_PacketSender* self){
     return self->slot_tail - self->slot_head;
 }
-static inline size_t pabb2_PacketSender_buffer_used(pabb2_PacketSender* self){
+static inline size_t pabb2_PacketSender_buffer_used(const pabb2_PacketSender* self){
     return self->buffer_tail - self->buffer_head;
 }
 

Common/PABotBase2/PABotbase2_ReliableStreamConnection.c

@@ -5,6 +5,7 @@
  */
 
 #include <stdio.h>  //  REMOVE
+#include "PABotBase2_ConnectionDebug.h"
 #include "PABotbase2_ReliableStreamConnection.h"
 
 
@@ -124,6 +125,7 @@ void pabb2_ReliableStreamConnection_run_events(pabb2_ReliableStreamConnection* s
             );
         }else{
             printf("Device: Failed to push.\n");
+            pabb2_StreamCoalescer_print(&self->stream_coalescer, true);
         }
         fflush(stdout);
         return;

SerialPrograms/Source/NintendoSwitch/DevPrograms/TestProgramComputer.cpp

@@ -281,7 +281,45 @@ struct RequestManagerConfig{
 
 
 
+std::string random_string(size_t max_length){
+    size_t length = rand() % max_length;
 
+    static const char TABLE[] = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
+
+    std::string ret;
+    for (size_t c = 0; c < length; c++){
+        ret += TABLE[rand() % 62 + 48];
+    }
+    return ret;
+}
+
+void stress_test(Logger& logger, CancellableScope& scope){
+    using namespace std::chrono_literals;
+
+    MockDevice device(GlobalThreadPools::unlimited_normal());
+
+    ReliableStreamConnection connection(
+        &scope,
+        logger, true,
+        GlobalThreadPools::unlimited_realtime(),
+        device,
+        100ms,
+        &device.print_lock()
+    );
+
+    //  Connect
+    connection.reset();
+    connection.send_request(PABB2_CONNECTION_OPCODE_ASK_VERSION);
+    connection.wait_for_pending();
+    connection.send_request(PABB2_CONNECTION_OPCODE_ASK_PACKET_SIZE);
+    connection.wait_for_pending();
+    connection.send_request(PABB2_CONNECTION_OPCODE_ASK_BUFFER_SLOTS);
+    connection.wait_for_pending();
+
+
+
+
+}
 
 
 
@@ -309,7 +347,7 @@ void TestProgramComputer::program(ProgramEnvironment& env, CancellableScope& sco
             logger, true,
             GlobalThreadPools::unlimited_realtime(),
             device,
-            1s,
+            100ms,
             &device.print_lock()
         );
 
@@ -330,16 +368,35 @@ void TestProgramComputer::program(ProgramEnvironment& env, CancellableScope& sco
         connection.send("zxcv", 4); device.push_expected_stream_data("zxcv", 4);
         connection.wait_for_pending();
 
-        cout << "sent = " << connection.send("0123456789abcdef", 16) << endl;
+        {
+            std::lock_guard<Mutex> lg(device.print_lock());
+            cout << "sent = " << connection.send("0123456789abcdef", 16) << endl;
+        }
+        device.push_expected_stream_data("0123456789abcdef", 16);
 //        connection.send("0123456789abcdef", 16);
 //        connection.send("0123456789abcdef", 16);
 //        connection.send("0123456789abcdef", 16);
 
-        connection.print();
-        device.print();
+        while (connection.pending() != 0 || device.verify_stream_data()){
+        }
 
         connection.wait_for_pending();
 
+        {
+            std::lock_guard<Mutex> lg(device.print_lock());
+            cout << "============> Done" << endl;
+        }
+        device.verify_stream_data();
+
+        connection.print();
+        device.print();
+
+        {
+            std::lock_guard<Mutex> lg(device.print_lock());
+            cout << "============> Finishing up" << endl;
+        }
+        scope.wait_for(5s);
+        device.verify_stream_data();
 
         scope.wait_for(60s);
     }