fixed bugs and added test for inference_handler.reset method

Author: faressc

docs/sphinx/architecture.rst

@@ -67,7 +67,7 @@ Component Responsibilities
 --------------------------
 
 :cpp:class:`anira::InferenceHandler`
-~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 The primary interface for users, handling the overall integration of neural network inference into audio processing workflows.
 
@@ -77,7 +77,7 @@ The primary interface for users, handling the overall integration of neural netw
 * Reports latency information
 
 :cpp:class:`anira::InferenceConfig`
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Stores configuration data for models and processing parameters.
 
@@ -87,7 +87,7 @@ Stores configuration data for models and processing parameters.
 * Memory management settings
 
 :cpp:class:`anira::PrePostProcessor`
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Handles data formatting between audio buffers and neural network tensors.
 
@@ -96,7 +96,7 @@ Handles data formatting between audio buffers and neural network tensors.
 * Manages intermediate buffers
 
 :cpp:class:`anira::InferenceManager`
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Coordinates the thread pool and inference scheduling.
 

include/anira/InferenceHandler.h

@@ -228,7 +228,7 @@ class ANIRA_API InferenceHandler {
      * @param channel Channel index to query (default: 0)
      * @return Number of samples received for the specified tensor and channel
      */
-    size_t get_num_received_samples(size_t tensor_index, size_t channel = 0) const;
+    size_t get_available_samples(size_t tensor_index, size_t channel = 0) const;
 
     /**
      * @brief Configures the handler for non-real-time operation

include/anira/scheduler/InferenceManager.h

@@ -162,7 +162,7 @@ class ANIRA_API InferenceManager {
      * @param channel Channel index to query
      * @return Number of samples received for the specified tensor and channel
      */
-    size_t get_num_received_samples(size_t tensor_index, size_t channel) const;
+    size_t get_available_samples(size_t tensor_index, size_t channel) const;
 
     /**
      * @brief Gets a const reference to the inference context (for unit testing)

src/InferenceHandler.cpp

@@ -129,8 +129,8 @@ std::vector<unsigned int> InferenceHandler::get_latency_vector() const {
     return m_inference_manager.get_latency();
 }
 
-size_t InferenceHandler::get_num_received_samples(size_t tensor_index, size_t channel) const {
-    return m_inference_manager.get_num_received_samples(tensor_index, channel);
+size_t InferenceHandler::get_available_samples(size_t tensor_index, size_t channel) const {
+    return m_inference_manager.get_available_samples(tensor_index, channel);
 }
 
 void InferenceHandler::set_non_realtime(bool is_non_realtime) {

src/benchmark/ProcessBlockFixture.cpp

@@ -13,7 +13,7 @@ ProcessBlockFixture::~ProcessBlockFixture() {
 }
 
 void ProcessBlockFixture::initialize_iteration() {
-    m_prev_num_received_samples = m_inference_handler->get_num_received_samples(0);
+    m_prev_num_received_samples = m_inference_handler->get_available_samples(0);
 }
 
 void ProcessBlockFixture::initialize_repetition(const InferenceConfig& inference_config, const HostConfig& host_config, const InferenceBackend& inference_backend, bool sleep_after_repetition) {
@@ -72,7 +72,7 @@ void ProcessBlockFixture::initialize_repetition(const InferenceConfig& inference
 }
 
 bool ProcessBlockFixture::buffer_processed() {
-    return m_inference_handler->get_num_received_samples(0) >= m_prev_num_received_samples;
+    return m_inference_handler->get_available_samples(0) >= m_prev_num_received_samples;
 }
 
 void ProcessBlockFixture::push_random_samples_in_buffer(anira::HostConfig host_config) {

src/scheduler/Context.cpp

@@ -93,7 +93,7 @@ void Context::release_thread_pool() {
 }
 
 void Context::release_session(std::shared_ptr<SessionElement> session) {
-    session->m_initialized.store(false, std::memory_order::acquire);
+    session->m_initialized.store(false, std::memory_order::release);
 
     drain_inference_queue(session);
 
@@ -134,7 +134,7 @@ void Context::release_session(std::shared_ptr<SessionElement> session) {
 }
 
 void Context::prepare_session(std::shared_ptr<SessionElement> session, HostConfig new_config, std::vector<long> custom_latency) {
-    session->m_initialized.store(false, std::memory_order::acquire);
+    session->m_initialized.store(false, std::memory_order::release);
 
     drain_inference_queue(session);
 
@@ -321,7 +321,7 @@ template <typename T> void Context::release_processor(InferenceConfig& inference
 }
 
 void Context::reset_session(std::shared_ptr<SessionElement> session) {
-    session->m_initialized.store(false, std::memory_order::acquire);
+    session->m_initialized.store(false, std::memory_order::release);
 
     drain_inference_queue(session);
 

src/scheduler/InferenceManager.cpp

@@ -147,9 +147,13 @@ const Context& InferenceManager::get_context() const {
     return *m_context;
 }
 
-size_t InferenceManager::get_num_received_samples(size_t tensor_index, size_t channel) const {
+size_t InferenceManager::get_available_samples(size_t tensor_index, size_t channel) const {
     m_context->new_data_request(m_session, 0.);
-    return m_session->m_receive_buffer[tensor_index].get_available_samples(channel);
+    if (m_inference_config.get_postprocess_output_size()[tensor_index] > 0) {
+        return m_session->m_receive_buffer[tensor_index].get_available_samples(channel);
+    } else {
+        return 0;
+    }
 }
 
 int InferenceManager::get_session_id() const {

test/test_InferenceHandler.cpp

@@ -128,13 +128,13 @@ TEST_P(InferenceTest, Simple){
             ring_buffer.push_sample(0, data_reference.at((repeat*buffer_size)+i));
         }
 
-        size_t prev_samples = inference_handler.get_num_received_samples(0);
+        size_t prev_samples = inference_handler.get_available_samples(0);
 
         inference_handler.process(test_buffer.get_array_of_write_pointers(), buffer_size);
 
         // wait until the block was properly processed
         auto start = std::chrono::system_clock::now();
-        while (inference_handler.get_num_received_samples(0) != prev_samples){
+        while (inference_handler.get_available_samples(0) != prev_samples){
             if (std::chrono::system_clock::now() >  start + std::chrono::duration<long int>(INFERENCE_TIMEOUT_S)){
                 FAIL() << "Timeout while waiting for block to be processed";
             }
@@ -215,13 +215,13 @@ TEST_P(InferenceTest, WithCustomLatency){
             ring_buffer.push_sample(0, data_reference.at((repeat*buffer_size)+i));
         }
 
-        size_t prev_samples = inference_handler.get_num_received_samples(0);
+        size_t prev_samples = inference_handler.get_available_samples(0);
 
         inference_handler.push_data(test_buffer.get_array_of_read_pointers(), buffer_size);
 
         // wait until the block was properly processed
         auto start = std::chrono::system_clock::now();
-        while (inference_handler.get_num_received_samples(0) != prev_samples + buffer_size){
+        while (inference_handler.get_available_samples(0) != prev_samples + buffer_size){
             if (std::chrono::system_clock::now() >  start + std::chrono::duration<long int>(INFERENCE_TIMEOUT_S)){
                 FAIL() << "Timeout while waiting for block to be processed";
             }
@@ -245,6 +245,137 @@ TEST_P(InferenceTest, WithCustomLatency){
     }
 }
 
+TEST_P(InferenceTest, Reset){
+
+    auto const& test_params = GetParam();
+    auto const& buffer_size = test_params.host_config.m_buffer_size;
+    auto const& reference_offset = test_params.reference_data_offset;
+
+    // read reference data
+    std::vector<float> data_input;
+    std::vector<float> data_reference;
+
+    read_wav(test_params.input_data_path, data_input);
+    read_wav(test_params.reference_data_path, data_reference);
+
+    ASSERT_TRUE(data_input.size() > 0);
+    ASSERT_TRUE(data_reference.size() > 0);
+
+    // setup inference
+    ContextConfig anira_context_config;
+    InferenceConfig inference_config = hybridnn_config;
+    HybridNNPrePostProcessor pp_processor(inference_config);
+    HybridNNBypassProcessor bypass_processor(inference_config);
+
+    // This test requires the buffer size to be a multiple of the preprocess input size
+    if (static_cast<size_t>(buffer_size) % inference_config.get_preprocess_input_size()[0] != 0){
+        GTEST_SKIP() << "Test requires the preprocess_input_size to be a multiple of the buffer size.";
+        return;
+    }
+
+    // Create an InferenceHandler instance
+    InferenceHandler inference_handler(pp_processor, inference_config, bypass_processor, anira_context_config);
+
+    // Allocate memory for audio processing
+    inference_handler.prepare(test_params.host_config);
+    // Select the inference backend
+    inference_handler.set_inference_backend(test_params.backend);
+
+    int latency_offset = inference_handler.get_latency(); // The 0th tensor is the audio data tensor, so we only need the first element of the latency vector
+
+    BufferF test_buffer(1, buffer_size);
+    RingBuffer ring_buffer;
+    ring_buffer.initialize_with_positions(1, latency_offset + buffer_size + reference_offset);
+    
+    //fill the buffer with zeroes to compensate for the latency
+    for (size_t i = 0; i < latency_offset + reference_offset; i++){
+        ring_buffer.push_sample(0, 0);
+    }    
+
+    // First, process some data to "contaminate" the internal state
+    for (size_t repeat = 0; repeat < 50; repeat++){
+        for (size_t i = 0; i < buffer_size; i++){
+            test_buffer.set_sample(0, i, data_input.at((repeat*buffer_size)+i));
+            ring_buffer.push_sample(0, data_reference.at((repeat*buffer_size)+i));
+        }
+        
+        size_t prev_samples = inference_handler.get_available_samples(0);
+        inference_handler.process(test_buffer.get_array_of_write_pointers(), buffer_size);
+        
+        // wait until the block was properly processed
+        auto start = std::chrono::system_clock::now();
+        while (inference_handler.get_available_samples(0) != prev_samples){
+            if (std::chrono::system_clock::now() >  start + std::chrono::duration<long int>(INFERENCE_TIMEOUT_S)){
+                FAIL() << "Timeout while waiting for block to be processed";
+            }
+            std::this_thread::sleep_for(std::chrono::nanoseconds (10));
+        }
+
+        for (size_t i = 0; i < buffer_size; i++){
+            float reference = ring_buffer.pop_sample(0);
+            float processed = test_buffer.get_sample(0, i);
+                        
+            if (repeat*buffer_size + i < latency_offset + reference_offset){
+                ASSERT_FLOAT_EQ(reference, 0);
+            } else {
+                // calculate epsilon on the fly
+                float epsilon = max(abs(reference), abs(processed)) * test_params.epsilon_rel + test_params.epsilon_abs; 
+                ASSERT_NEAR(reference, processed, epsilon) << "repeat=" << repeat << ", i=" << i << ", total sample nr: " << repeat*buffer_size + i  << std::endl;
+            }
+        }
+    }
+
+    // Now reset the inference handler
+    inference_handler.reset();
+
+    // Verify that the available samples count is reset
+    EXPECT_EQ(inference_handler.get_available_samples(0), latency_offset) << "Available samples should be " << latency_offset << " after reset";
+
+    // Reset the ring buffer to restart from the beginning of reference data
+    ring_buffer.clear_with_positions();
+    ring_buffer.initialize_with_positions(1, latency_offset + buffer_size + reference_offset);
+    
+    // Fill the buffer with zeroes to compensate for the latency
+    for (size_t i = 0; i < latency_offset + reference_offset; i++){
+        ring_buffer.push_sample(0, 0);
+    }
+
+    // Process data again and verify that output matches reference from the beginning
+    for (size_t repeat = 0; repeat < 150; repeat++){
+
+        for (size_t i = 0; i < buffer_size; i++){
+            test_buffer.set_sample(0, i, data_input.at((repeat*buffer_size)+i));
+            ring_buffer.push_sample(0, data_reference.at((repeat*buffer_size)+i));
+        }
+        
+        size_t prev_samples = inference_handler.get_available_samples(0);
+
+        inference_handler.process(test_buffer.get_array_of_write_pointers(), buffer_size);
+        
+        // wait until the block was properly processed
+        auto start = std::chrono::system_clock::now();
+        while (inference_handler.get_available_samples(0) != prev_samples){
+            if (std::chrono::system_clock::now() >  start + std::chrono::duration<long int>(INFERENCE_TIMEOUT_S)){
+                FAIL() << "Timeout while waiting for block to be processed";
+            }
+            std::this_thread::sleep_for(std::chrono::nanoseconds (10));
+        }        
+
+        for (size_t i = 0; i < buffer_size; i++){
+            float reference = ring_buffer.pop_sample(0);
+            float processed = test_buffer.get_sample(0, i);
+                        
+            if (repeat*buffer_size + i < latency_offset + reference_offset){
+                ASSERT_FLOAT_EQ(reference, 0);
+            } else {
+                // calculate epsilon on the fly
+                float epsilon = max(abs(reference), abs(processed)) * test_params.epsilon_rel + test_params.epsilon_abs; 
+                ASSERT_NEAR(reference, processed, epsilon) << "After reset: repeat=" << repeat << ", i=" << i << ", total sample nr: " << repeat*buffer_size + i  << std::endl;
+            }
+        }
+    }
+}
+
 std::string build_test_name(const testing::TestParamInfo<InferenceTest::ParamType>& info){
     std::stringstream ss_sample_rate, ss_buffer_size;