Intermittent producer test scenarios (#584)

MushMal · web-flow · commit 017ab505f072 · 2020-10-09T09:53:46.000-04:00
* Intermittent producer test scenarios

* Due to timing, removing the checks for the in-order buffering ACK check. This might fail in case of slight timing on a sessions edge

* Fixing a build issue with cmake related to Policy CMP0054

* Fixing a build issue with cmake related to Policy CMP0054

* Updating C producer to resolve the CMake compile compatibility issue

* Trying to fix windows build
diff --git a/.gitignore b/.gitignore
@@ -13,6 +13,7 @@ open-source/libsrtp
 open-source/libusrsctp
 open-source/libwebsockets
 open-source/local
+open-source/libautoconf
 outputs
 tags
 dependency
diff --git a/.travis.yml b/.travis.yml
@@ -107,7 +107,8 @@ matrix:
     - name: "Windows MSVC"
       os: windows
       script:
-        - choco install nasm strawberryperl gstreamer
+        - choco install nasm strawberryperl
+        - choco install gstreamer --version=1.16.2
         - choco install gstreamer-devel --version=1.16.2 # gstreamer-devel has not been approved yet. Version number must be explicit to install
         - unset CC CC_FOR_BUILD CXX CXX_FOR_BUILD # We want to use MSVC
         - export "PATH=/c/Strawberry/perl/site/bin:/c/Strawberry/perl/bin:/c/Strawberry/c/bin:/c/Program Files/NASM:`pwd`/open-source/local/lib:`pwd`/open-source/local/bin:$PATH"
diff --git a/CMake/Dependencies/libkvscproducer-CMakeLists.txt b/CMake/Dependencies/libkvscproducer-CMakeLists.txt
@@ -7,7 +7,7 @@ include(ExternalProject)
 # clone repo only
 ExternalProject_Add(libkvscproducer-download
 	GIT_REPOSITORY    https://github.com/awslabs/amazon-kinesis-video-streams-producer-c.git
-    GIT_TAG           ca3401496223bd536faff4eaa8f14eb01c0642f9
+    GIT_TAG           22edebfd87f5a38ab8af58da9a42f3d8dc7aebe7
     SOURCE_DIR        "${CMAKE_CURRENT_BINARY_DIR}/kvscproducer-src"
 	BINARY_DIR        "${CMAKE_CURRENT_BINARY_DIR}/kvscproducer-build"
 	CONFIGURE_COMMAND ""
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -39,7 +39,7 @@ else()
   set(LINKAGE SHARED)
 endif()
 
-set(KINESIS_VIDEO_PRODUCER_CPP_SRC "${CMAKE_CURRENT_SOURCE_DIR}")
+set(KINESIS_VIDEO_PRODUCER_CPP_SRC ${CMAKE_CURRENT_SOURCE_DIR})
 set(KINESIS_VIDEO_OPEN_SOURCE_SRC ${CMAKE_CURRENT_SOURCE_DIR}/open-source)
 
 message(STATUS "Kinesis Video Cpp Producer path is ${KINESIS_VIDEO_PRODUCER_CPP_SRC}")
@@ -121,7 +121,7 @@ if (WIN32)
 endif()
 
 ############# Enable Sanitizers ############
-if("${CMAKE_C_COMPILER_ID}" MATCHES "GNU|Clang")
+if(${CMAKE_C_COMPILER_ID} MATCHES "GNU|Clang")
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC")
 
   if(ADD_MUCLIBC)
diff --git a/tst/ProducerFunctionalityTest.cpp b/tst/ProducerFunctionalityTest.cpp
@@ -325,6 +325,258 @@ TEST_F(ProducerFunctionalityTest, offline_mode_token_rotation_block_on_space) {
 
 }
 
+/**
+ * Set short max latency in the stream info.
+ * Send a few fragments then pause as in an intermittent
+ * scenario. Put an EoFR before pausing to ensure we terminate the fragment on the backend
+ * which will issue a persistent ACK, causing the state machine to not rollback
+ * on the next frame produced after the pause. The pause will cause the connection timeout
+ * which would then set the state machine to change state to issue a new session. The new
+ * session will not rollback due to the previous session being closed and persisted ACK
+ * received. If, however, this is not the case, the rollback would cause a latency pressure.
+ */
+TEST_F(ProducerFunctionalityTest, realtime_intermittent_no_latency_pressure_eofr) {
+    // Check if it's run with the env vars set if not bail out
+    if (!access_key_set_) {
+        return;
+    }
+
+    CreateProducer();
+
+    buffering_ack_in_sequence_ = true;
+    key_frame_interval_ = 60;
+    total_frame_count_ = 6 * key_frame_interval_;
+
+    UINT64 timestamp = 0;
+    Frame frame;
+    frame.duration = 16 * HUNDREDS_OF_NANOS_IN_A_MILLISECOND;
+    frame.frameData = frameBuffer_;
+    frame.size = SIZEOF(frameBuffer_);
+    frame.trackId = DEFAULT_TRACK_ID;
+    Frame eofr = EOFR_FRAME_INITIALIZER;
+
+    // Set the value of the frame buffer
+    MEMSET(frame.frameData, 0x55, SIZEOF(frameBuffer_));
+
+    streams_[0] = CreateTestStream(0, STREAMING_TYPE_REALTIME,
+                                   15000,
+                                   120);
+    auto kinesis_video_stream = streams_[0];
+
+    for(uint32_t i = 0; i < total_frame_count_; i++) {
+        frame.index = i;
+        frame.flags = (frame.index % key_frame_interval_ == 0) ? FRAME_FLAG_KEY_FRAME : FRAME_FLAG_NONE;
+
+        // Pause on the 5th
+        if (i == 5 * key_frame_interval_) {
+            EXPECT_TRUE(kinesis_video_stream->putFrame(eofr));
+
+            // Make sure we hit the connection idle timeout
+            THREAD_SLEEP(60 * HUNDREDS_OF_NANOS_IN_A_SECOND);
+        }
+
+        timestamp = GETTIME();
+        frame.decodingTs = timestamp;
+        frame.presentationTs = timestamp;
+
+        std::stringstream strstrm;
+        strstrm << " TID: 0x" << std::hex << GETTID();
+        LOG_INFO("Putting frame for stream: " << kinesis_video_stream->getStreamName()
+                                              << strstrm.str()
+                                              << " Id: " << frame.index
+                                              << ", Key Frame: "
+                                              << (((frame.flags & FRAME_FLAG_KEY_FRAME) == FRAME_FLAG_KEY_FRAME)
+                                                  ? "true" : "false")
+                                              << ", Size: " << frame.size
+                                              << ", Dts: " << frame.decodingTs
+                                              << ", Pts: " << frame.presentationTs);
+
+        EXPECT_TRUE(kinesis_video_stream->putFrame(frame));
+        timestamp += frame_duration_;
+
+        THREAD_SLEEP(30 * HUNDREDS_OF_NANOS_IN_A_MILLISECOND);
+    }
+
+    THREAD_SLEEP(WAIT_5_SECONDS_FOR_ACKS);
+    LOG_DEBUG("Stopping the stream: " << kinesis_video_stream->getStreamName());
+    EXPECT_TRUE(kinesis_video_stream->stopSync()) << "Timed out awaiting for the stream stop notification";
+    EXPECT_FALSE(frame_dropped_) << "Status of frame drop " << frame_dropped_;
+    EXPECT_EQ(0, latency_pressure_count_) << "Should not have latency pressure events";
+    EXPECT_TRUE(STATUS_SUCCEEDED(getErrorStatus())) << "Status of stream error " << getErrorStatus();
+    kinesis_video_producer_->freeStreams();
+    streams_[0] = nullptr;
+}
+
+/**
+ * Set short max latency in the stream info.
+ * Set an automatic intermittent producer case handling (intra-frame timeout/closing fragment).
+ * Send a few fragments then pause as in an intermittent
+ * scenario. No EoFR produced before pausing.
+ * Automatic fragment closing fired on timeout to ensure we terminate the fragment on the backend
+ * which will issue a persistent ACK, causing the state machine to not rollback
+ * on the next frame produced after the pause. The pause will cause the connection timeout
+ * which would then set the state machine to change state to issue a new session. The new
+ * session will not rollback due to the previous session being closed and persisted ACK
+ * received. If, however, this is not the case, the rollback would cause a latency pressure.
+ */
+ /**
+  * TODO: This is disabled until we implement the auto fragment closing feature
+  */
+TEST_F(ProducerFunctionalityTest, DISABLED_realtime_intermittent_no_latency_pressure_auto) {
+    // Check if it's run with the env vars set if not bail out
+    if (!access_key_set_) {
+        return;
+    }
+
+    CreateProducer();
+
+    buffering_ack_in_sequence_ = true;
+    key_frame_interval_ = 60;
+    total_frame_count_ = 6 * key_frame_interval_;
+
+    UINT64 timestamp = 0;
+    Frame frame;
+    frame.duration = 16 * HUNDREDS_OF_NANOS_IN_A_MILLISECOND;
+    frame.frameData = frameBuffer_;
+    frame.size = SIZEOF(frameBuffer_);
+    frame.trackId = DEFAULT_TRACK_ID;
+
+    // Set the value of the frame buffer
+    MEMSET(frame.frameData, 0x55, SIZEOF(frameBuffer_));
+
+    streams_[0] = CreateTestStream(0, STREAMING_TYPE_REALTIME,
+                                   15000,
+                                   120);
+    auto kinesis_video_stream = streams_[0];
+
+    for(uint32_t i = 0; i < total_frame_count_; i++) {
+        frame.index = i;
+        frame.flags = (frame.index % key_frame_interval_ == 0) ? FRAME_FLAG_KEY_FRAME : FRAME_FLAG_NONE;
+
+        // Pause on the 5th
+        if (i == 5 * key_frame_interval_) {
+            // Make sure we hit the connection idle timeout
+            THREAD_SLEEP(60 * HUNDREDS_OF_NANOS_IN_A_SECOND);
+        }
+
+        timestamp = GETTIME();
+        frame.decodingTs = timestamp;
+        frame.presentationTs = timestamp;
+
+        std::stringstream strstrm;
+        strstrm << " TID: 0x" << std::hex << GETTID();
+        LOG_INFO("Putting frame for stream: " << kinesis_video_stream->getStreamName()
+                                              << strstrm.str()
+                                              << " Id: " << frame.index
+                                              << ", Key Frame: "
+                                              << (((frame.flags & FRAME_FLAG_KEY_FRAME) == FRAME_FLAG_KEY_FRAME)
+                                                  ? "true" : "false")
+                                              << ", Size: " << frame.size
+                                              << ", Dts: " << frame.decodingTs
+                                              << ", Pts: " << frame.presentationTs);
+
+        EXPECT_TRUE(kinesis_video_stream->putFrame(frame));
+        timestamp += frame_duration_;
+
+        THREAD_SLEEP(30 * HUNDREDS_OF_NANOS_IN_A_MILLISECOND);
+    }
+
+    THREAD_SLEEP(WAIT_5_SECONDS_FOR_ACKS);
+    LOG_DEBUG("Stopping the stream: " << kinesis_video_stream->getStreamName());
+    EXPECT_TRUE(kinesis_video_stream->stopSync()) << "Timed out awaiting for the stream stop notification";
+    EXPECT_FALSE(frame_dropped_) << "Status of frame drop " << frame_dropped_;
+    EXPECT_EQ(0, latency_pressure_count_) << "Should not have latency pressure events";
+    EXPECT_TRUE(STATUS_SUCCEEDED(getErrorStatus())) << "Status of stream error " << getErrorStatus();
+    EXPECT_TRUE(buffering_ack_in_sequence_); // all fragments should be sent
+    kinesis_video_producer_->freeStreams();
+    streams_[0] = nullptr;
+}
+
+/**
+ * Set short max latency in the stream info.
+ * Send a few fragments then pause as in an intermittent
+ * scenario. No EoFR produced before pausing.
+ * The backend doesn't terminate the fragment so no ACKs will be issued.
+ * The pause will cause the connection timeout
+ * which would then set the state machine to change state to issue a new session. The new
+ * session will not rollback due to the previous session being closed and persisted ACK
+ * received.
+ * As we haven't received a persistent ACK, the rollback will roll it back to the previous
+ * fragment and re-stream. The previous fragment timestamps are in the past causing latency
+ * pressure callback to fire.
+ */
+TEST_F(ProducerFunctionalityTest, realtime_intermittent_latency_pressure) {
+    // Check if it's run with the env vars set if not bail out
+    if (!access_key_set_) {
+        return;
+    }
+
+    KinesisVideoLogger::getInstance().setLogLevel(log4cplus::DEBUG_LOG_LEVEL);
+
+    CreateProducer();
+
+    buffering_ack_in_sequence_ = true;
+    key_frame_interval_ = 60;
+    total_frame_count_ = 6 * key_frame_interval_;
+
+    UINT64 timestamp = 0;
+    Frame frame;
+    frame.duration = 16 * HUNDREDS_OF_NANOS_IN_A_MILLISECOND;
+    frame.frameData = frameBuffer_;
+    frame.size = SIZEOF(frameBuffer_);
+    frame.trackId = DEFAULT_TRACK_ID;
+
+    // Set the value of the frame buffer
+    MEMSET(frame.frameData, 0x55, SIZEOF(frameBuffer_));
+
+    streams_[0] = CreateTestStream(0, STREAMING_TYPE_REALTIME,
+                                   15000,
+                                   120);
+    auto kinesis_video_stream = streams_[0];
+
+    for(uint32_t i = 0; i < total_frame_count_; i++) {
+        frame.index = i;
+        frame.flags = (frame.index % key_frame_interval_ == 0) ? FRAME_FLAG_KEY_FRAME : FRAME_FLAG_NONE;
+
+        // Pause on the 5th
+        if (i == 5 * key_frame_interval_) {
+            // Make sure we hit the connection idle timeout
+            THREAD_SLEEP(60 * HUNDREDS_OF_NANOS_IN_A_SECOND);
+        }
+
+        timestamp = GETTIME();
+        frame.decodingTs = timestamp;
+        frame.presentationTs = timestamp;
+
+        std::stringstream strstrm;
+        strstrm << " TID: 0x" << std::hex << GETTID();
+        LOG_INFO("Putting frame for stream: " << kinesis_video_stream->getStreamName()
+                                              << strstrm.str()
+                                              << " Id: " << frame.index
+                                              << ", Key Frame: "
+                                              << (((frame.flags & FRAME_FLAG_KEY_FRAME) == FRAME_FLAG_KEY_FRAME)
+                                                  ? "true" : "false")
+                                              << ", Size: " << frame.size
+                                              << ", Dts: " << frame.decodingTs
+                                              << ", Pts: " << frame.presentationTs);
+
+        EXPECT_TRUE(kinesis_video_stream->putFrame(frame));
+        timestamp += frame_duration_;
+
+        THREAD_SLEEP(30 * HUNDREDS_OF_NANOS_IN_A_MILLISECOND);
+    }
+
+    THREAD_SLEEP(WAIT_5_SECONDS_FOR_ACKS);
+    LOG_DEBUG("Stopping the stream: " << kinesis_video_stream->getStreamName());
+    EXPECT_TRUE(kinesis_video_stream->stopSync()) << "Timed out awaiting for the stream stop notification";
+    EXPECT_FALSE(frame_dropped_) << "Status of frame drop " << frame_dropped_;
+    EXPECT_NE(0, latency_pressure_count_) << "Should fire latency pressure events";
+    EXPECT_TRUE(STATUS_SUCCEEDED(getErrorStatus())) << "Status of stream error " << getErrorStatus();
+    EXPECT_FALSE(buffering_ack_in_sequence_);
+    kinesis_video_producer_->freeStreams();
+    streams_[0] = nullptr;
+}
+
 }
 }
 }
diff --git a/tst/ProducerTestFixture.cpp b/tst/ProducerTestFixture.cpp
@@ -70,6 +70,15 @@ STATUS TestStreamCallbackProvider::droppedFrameReportHandler(UINT64 custom_data,
 STATUS TestStreamCallbackProvider::streamLatencyPressureHandler(UINT64 custom_data, STREAM_HANDLE stream_handle, UINT64 duration) {
     UNUSED_PARAM(stream_handle);
     LOG_WARN("Reporting stream latency pressure. Current buffer duration " << duration);
+
+    ProducerTestBase* testBase;
+    STATUS ret = getProducerTestBase(custom_data, &testBase);
+    if (STATUS_FAILED(ret)) {
+        return ret;
+    }
+
+    testBase->latency_pressure_count_++;
+
     return validateCallback(custom_data);
 }
 
diff --git a/tst/ProducerTestFixture.h b/tst/ProducerTestFixture.h
@@ -180,6 +180,7 @@ class ProducerTestBase : public ::testing::Test {
                          defaultRegion_(DEFAULT_AWS_REGION),
                          caCertPath_(""),
                          error_status_(STATUS_SUCCESS),
+                         latency_pressure_count_(0),
                          device_storage_size_(TEST_STORAGE_SIZE_IN_BYTES),
                          fps_(TEST_FPS),
                          total_frame_count_(TEST_TOTAL_FRAME_COUNT),
@@ -228,6 +229,7 @@ class ProducerTestBase : public ::testing::Test {
     std::atomic_bool storage_overflow_;
     std::atomic_bool buffering_ack_in_sequence_;
     std::atomic_uint error_status_;
+    std::atomic_uint latency_pressure_count_;
     std::map<UPLOAD_HANDLE, uint64_t> previous_buffering_ack_timestamp_;
 
 protected:
