Merge 889af3194d into 6141d27841

Handle reliable stream send failures

CMakeLists.txt

@@ -57,6 +57,7 @@ set(UCLIENT_MAX_INPUT_RELIABLE_STREAMS 1 CACHE STRING "Set the maximum number of
 set(UCLIENT_MAX_SESSION_CONNECTION_ATTEMPTS 10 CACHE STRING "Set the number of connection attemps.")
 set(UCLIENT_MIN_SESSION_CONNECTION_INTERVAL 1000 CACHE STRING "Set the connection interval in milliseconds.")
 set(UCLIENT_MIN_HEARTBEAT_TIME_INTERVAL 100 CACHE STRING "Set the time interval between heartbeats in milliseconds.")
+set(UCLIENT_RELIABLE_RESENT_TIME 100 CACHE STRING "Set the time for reliable stream resend in milliseconds.")
 set(UCLIENT_UDP_TRANSPORT_MTU 512 CACHE STRING "Set the UDP transport MTU.")
 set(UCLIENT_TCP_TRANSPORT_MTU 512 CACHE STRING "Set the TCP transport MTU.")
 set(UCLIENT_SERIAL_TRANSPORT_MTU 512 CACHE STRING "Set the Serial transport MTU.")

include/uxr/client/config.h.in

@@ -46,6 +46,7 @@
 #define UXR_CONFIG_MAX_SESSION_CONNECTION_ATTEMPTS    @UCLIENT_MAX_SESSION_CONNECTION_ATTEMPTS@
 #define UXR_CONFIG_MIN_SESSION_CONNECTION_INTERVAL    @UCLIENT_MIN_SESSION_CONNECTION_INTERVAL@
 #define UXR_CONFIG_MIN_HEARTBEAT_TIME_INTERVAL        @UCLIENT_MIN_HEARTBEAT_TIME_INTERVAL@
+#define UXR_CONFIG_RELIABLE_RESENT_TIME               @UCLIENT_RELIABLE_RESENT_TIME@
 
 #ifdef UCLIENT_PROFILE_UDP
 #define UXR_CONFIG_UDP_TRANSPORT_MTU                  @UCLIENT_UDP_TRANSPORT_MTU@

src/c/core/session/session.c

@@ -629,7 +629,11 @@ void uxr_flash_output_streams(
         while (uxr_prepare_next_reliable_buffer_to_send(stream, &buffer, &length, &seq_num))
         {
             uxr_stamp_session_header(&session->info, id.raw, seq_num, buffer);
-            send_message(session, buffer, length);
+            if (!send_message(session, buffer, length))
+            {
+                uxr_cancel_reliable_buffer_to_send(stream, seq_num);
+                break;
+            }
         }
 
         UXR_UNLOCK_STREAM_ID(session, id);

src/c/core/session/stream/output_reliable_stream.c

@@ -9,6 +9,7 @@
 #include "./common_reliable_stream_internal.h"
 #include "../submessage_internal.h"
 #include <uxr/client/profile/multithread/multithread.h>
+#include <uxr/client/util/time.h>
 
 #define MIN_HEARTBEAT_TIME_INTERVAL ((int64_t) UXR_CONFIG_MIN_HEARTBEAT_TIME_INTERVAL) // ms
 
@@ -64,6 +65,7 @@ bool uxr_prepare_reliable_buffer_to_write(
 
     uint16_t available_block_size = (uint16_t)(buffer_capacity - (uint16_t)(stream->offset + SUBHEADER_SIZE));
     size_t remaining_blocks = get_available_free_slots(stream);
+    static int64_t reliable_ackend_ts = 0x0;
 
     // Aligment required for inserting an XRCE subheader
     buffer_size += ucdr_alignment(buffer_size, 4);
@@ -164,6 +166,17 @@ bool uxr_prepare_reliable_buffer_to_write(
         }
     }
 
+    if (available_to_write == false) {
+        if (reliable_ackend_ts == 0x0) {
+            reliable_ackend_ts = uxr_millis() + UXR_CONFIG_RELIABLE_RESENT_TIME;
+        } else if (reliable_ackend_ts <= uxr_millis()) {
+            stream->last_sent = uxr_seq_num_sub(stream->last_sent, 1);
+            reliable_ackend_ts = uxr_millis() + UXR_CONFIG_RELIABLE_RESENT_TIME;
+        }
+    } else {
+        reliable_ackend_ts = 0x0;
+    }
+
     return available_to_write;
 }
 
@@ -192,6 +205,19 @@ bool uxr_prepare_next_reliable_buffer_to_send(
     return data_to_send;
 }
 
+void uxr_cancel_reliable_buffer_to_send(
+        uxrOutputReliableStream* stream,
+        uxrSeqNum seq_num)
+{
+    uxrSeqNum next_seq_num = uxr_seq_num_add(seq_num, 1);
+    stream->last_sent = uxr_seq_num_sub(seq_num, 1);
+    if (stream->last_written == next_seq_num &&
+            stream->offset == uxr_get_reliable_buffer_size(&stream->base, next_seq_num))
+    {
+        stream->last_written = seq_num;
+    }
+}
+
 bool uxr_update_output_stream_heartbeat_timestamp(
         uxrOutputReliableStream* stream,
         int64_t current_timestamp)

src/c/core/session/stream/output_reliable_stream_internal.h

@@ -45,6 +45,9 @@ bool uxr_prepare_next_reliable_buffer_to_send(
         uint8_t** buffer,
         size_t* length,
         uxrSeqNum* seq_num);
+void uxr_cancel_reliable_buffer_to_send(
+        uxrOutputReliableStream* stream,
+        uxrSeqNum seq_num);
 
 bool uxr_update_output_stream_heartbeat_timestamp(
         uxrOutputReliableStream* stream,

test/shared_memory/CMakeLists.txt

@@ -78,3 +78,5 @@ set_target_properties(sharedmem_test PROPERTIES
     CXX_STANDARD_REQUIRED
         YES
     )
+
+include_directories(SYSTEM ${PROJECT_SOURCE_DIR}/../../src)

test/shared_memory/SharedMemory.cpp

@@ -1,3 +1,8 @@
+extern "C"
+{
+#include <c/util/time.c>
+}
+
 #include <gtest/gtest.h>
 #include <uxr/client/client.h>
 

test/unitary/session/Session.cpp

@@ -103,6 +103,7 @@ public:
     uint8_t input_reliable_buffer[MTU * HISTORY];
 
     static int listening_counter;
+    static int sending_counter;
 
     static bool send_msg(
             void* instance,
@@ -124,6 +125,13 @@ public:
             EXPECT_EQ(size_t(MTU), len);
             return false;
         }
+        else if (std::string("FlashReliableStreamSendError") ==
+                ::testing::UnitTest::GetInstance()->current_test_info()->name())
+        {
+            EXPECT_EQ(size_t(OFFSET + SUBHEADER_SIZE + 8), len);
+            SessionTest::sending_counter++;
+            return false;
+        }
         else if (std::string("SendHeartbeat") == ::testing::UnitTest::GetInstance()->current_test_info()->name())
         {
             EXPECT_EQ(size_t(HEARTBEAT_MAX_MSG_SIZE - (MAX_HEADER_SIZE - OFFSET)), len);
@@ -272,6 +280,7 @@ public:
 
 SessionTest* SessionTest::current = nullptr;
 int SessionTest::listening_counter;
+int SessionTest::sending_counter;
 
 TEST_F(SessionTest, SetStatusCallback)
 {
@@ -355,6 +364,28 @@ TEST_F(SessionTest, FlashStreams)
     uxr_flash_output_streams(&session);
 }
 
+TEST_F(SessionTest, FlashReliableStreamSendError)
+{
+    SessionTest::sending_counter = 0;
+
+    ucdrBuffer ub;
+    uxrStreamId output_reliable = uxr_stream_id(0, UXR_RELIABLE_STREAM, UXR_OUTPUT_STREAM);
+    (void) uxr_prepare_stream_to_write_submessage(&session, output_reliable, 8, &ub, 1, 0);
+
+    uxrOutputReliableStream* stream = &session.streams.output_reliable[0];
+    uxr_flash_output_streams(&session);
+
+    EXPECT_EQ(1, SessionTest::sending_counter);
+    EXPECT_EQ(SEQ_NUM_MAX, stream->last_sent);
+    EXPECT_EQ(0u, stream->last_written);
+
+    uxr_flash_output_streams(&session);
+
+    EXPECT_EQ(2, SessionTest::sending_counter);
+    EXPECT_EQ(SEQ_NUM_MAX, stream->last_sent);
+    EXPECT_EQ(0u, stream->last_written);
+}
+
 TEST_F(SessionTest, WaitSessionStatusBad)
 {
     // The OK version is already checked with the CreateOk and DeleteOk test versions

test/unitary/session/streams/OutputReliableStream.cpp

@@ -8,6 +8,7 @@ extern "C"
 #include <c/core/serialization/xrce_subheader.c>
 #include <c/core/session/submessage.c>
 #include <c/core/serialization/xrce_header.c>
+#include <c/util/time.c>
 }
 
 #define BUFFER_SIZE           size_t(128)