Changed queue implementation to the proposed vector-based solution.

webrtc/modules/audio_processing/swap_queue_unittest.cc

+/*
+ *  Copyright (c) 2015 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#include <algorithm>
+#include <tuple>
+#include <vector>
+
+#include "testing/gtest/include/gtest/gtest.h"
+#include "webrtc/modules/audio_processing/swap_queue.h"
+#include "webrtc/modules/interface/module_common_types.h"
+#include "webrtc/system_wrappers/interface/event_wrapper.h"
+#include "webrtc/system_wrappers/interface/sleep.h"
+#include "webrtc/system_wrappers/interface/thread_wrapper.h"
+
+namespace webrtc {
+
+namespace {
+
+// The type for a message.
+class SwapQueueTestMessage {
+ public:
+  SwapQueueTestMessage() {
+    id_ = -1;
+    info_ = -1;
+  }
+
+  SwapQueueTestMessage(const SwapQueueTestMessage& m)
+      : id_(m.id_), info_(m.info_) {}
+
+  explicit SwapQueueTestMessage(int id, int info) {
+    id_ = id;
+    info_ = info;
+  }
+
+  bool Compare(const SwapQueueTestMessage& m) {
+    return ((id_ == m.id_) && (info_ == m.info_));
+  }
+
+  void swap(SwapQueueTestMessage& m) {
+    std::swap(id_, m.id_);
+    std::swap(info_, m.info_);
+  }
+
+ private:
+  int id_;
+  int info_;
+};
+
+// Sample data type enum.
+enum class DataType { Float, Int16 };
+
+// Implements the tests for the sample queue.
+class SwapQueueMessageTest
+    : public ::testing::TestWithParam<::testing::tuple<size_t, size_t>> {
+ public:
+  SwapQueueMessageTest()
+      : test_complete_(EventWrapper::Create()),
+        reader_thread_(
+            ThreadWrapper::CreateThread(CbReaderThread, this, "reader")),
+        writer_thread_(
+            ThreadWrapper::CreateThread(CbWriterThread, this, "writer")) {}
+
+  // Run the test with a timeout.
+  EventTypeWrapper RunTest() {
+    StartThreads();
+    return test_complete_->Wait(kTestTimeOutLimit);
+  }
+
+  virtual void SetUp() {
+    // Read test parameter
+    size_t queue_size = static_cast<size_t>(testing::get<0>(GetParam()));
+    size_t messages_size = static_cast<size_t>(testing::get<1>(GetParam()));
+
+    // Create queues.
+    queue_.reset(new SwapQueue<SwapQueueTestMessage>(queue_size));
+
+    // Create and populate message vector.
+    messages_ = std::vector<SwapQueueTestMessage>();
+    CreateTestData(messages_size);
+  }
+
+  virtual void TearDown() {
+    reader_thread_->Stop();
+    writer_thread_->Stop();
+  }
+
+ private:
+  const int kTestTimeOutLimit = 10 * 1000;
+
+  // Populates the data vectors with random values.
+  void CreateTestData(int messages_size) {
+    messages_.resize(messages_size);
+    for (size_t k = 0; k < messages_.size(); k++)
+      messages_[k] = SwapQueueTestMessage(rand_r(&writer_seed) % 10000,
+                                          rand_r(&writer_seed) % 10000);
+  }
+
+  // Thread callback for the reader thread
+  static bool CbReaderThread(void* context) {
+    return reinterpret_cast<SwapQueueMessageTest*>(context)->CbReaderImpl();
+  }
+
+  // Thread callback for the writer thread
+  static bool CbWriterThread(void* context) {
+    return reinterpret_cast<SwapQueueMessageTest*>(context)->CbWriterImpl();
+  }
+
+  // Tests in a threadsafe manner whether all the samples have been
+  // written and read.
+  bool TestDone() { return (GetNumMessagesRead() == messages_.size()); }
+
+  // Returns in a threadsafe manner the number of frames read from the queue.
+  size_t GetNumMessagesRead() {
+    rtc::CritScope cs(&crit_);
+    return num_messages_read_;
+  }
+
+  // Increases in a threadsafe manner the number of messages
+  // read from the queue.
+  void IncreaseMessageReadCounter() {
+    rtc::CritScope cs(&crit_);
+    num_messages_read_++;
+  }
+
+  // Implements the callback functionality for the reader thread.
+  bool CbReaderImpl() {
+    SleepRandomTime(3, &reader_seed);
+
+    // Read the message and verify bitexactness.
+    size_t num_messages_read = GetNumMessagesRead();
+    SwapQueueTestMessage m;
+    if (queue_->Remove(&m)) {
+      EXPECT_TRUE(m.Compare(messages_[num_messages_read]));
+      IncreaseMessageReadCounter();
+    }
+
+    // End the test if the test is done or an assert has occurred.
+    if (TestDone() || HasFatalFailure()) {
+      test_complete_->Set();
+    }
+
+    return true;
+  }
+
+  // Implements the callback functionality for the writer thread.
+  bool CbWriterImpl() {
+    SleepRandomTime(3, &writer_seed);
+
+    if (num_messages_written_ < messages_.size()) {
+      // Attempt to put the message in the queue.
+      SwapQueueTestMessage m(messages_[num_messages_written_]);
+      if (queue_->Insert(&m))
+        num_messages_written_++;
+    }
+
+    // End the test early if a fatal failure (ASSERT_*) has occurred.
+    if (HasFatalFailure())
+      test_complete_->Set();
+
+    return true;
+  }
+
+  // Sleeps a random time.
+  static void SleepRandomTime(int max_sleep, unsigned int* seed) {
+    SleepMs(rand_r(seed) % (max_sleep + 1));
+  }
+
+  // Start the threads used in the test.
+  void StartThreads() {
+    EXPECT_TRUE(reader_thread_->Start());
+    reader_thread_->SetPriority(kRealtimePriority);
+    EXPECT_TRUE(writer_thread_->Start());
+    writer_thread_->SetPriority(kRealtimePriority);
+  }
+
+  // Test parameters.
+  rtc::scoped_ptr<SwapQueue<SwapQueueTestMessage>> queue_;
+  rtc::CriticalSection crit_;
+  const rtc::scoped_ptr<EventWrapper> test_complete_;
+  rtc::scoped_ptr<ThreadWrapper> reader_thread_;
+  rtc::scoped_ptr<ThreadWrapper> writer_thread_;
+  unsigned int reader_seed = 42;
+  unsigned int writer_seed = 37;
+  std::vector<SwapQueueTestMessage> messages_;
+  std::vector<int16_t> buffer_writer_int16_;
+  size_t num_messages_written_ = 0;
+  size_t num_messages_read_ GUARDED_BY(crit_) = 0;
+};
+
+// Implements the tests sample queue.
+class SwapQueueSampleTest
+    : public ::testing::TestWithParam<
+          ::testing::tuple<size_t, size_t, size_t, DataType>> {
+ public:
+  SwapQueueSampleTest()
+      : test_complete_(EventWrapper::Create()),
+        reader_thread_(
+            ThreadWrapper::CreateThread(CbReaderThread, this, "reader")),
+        writer_thread_(
+            ThreadWrapper::CreateThread(CbWriterThread, this, "writer")) {}
+
+  // Run the test with a timeout.
+  EventTypeWrapper RunTest() {
+    StartThreads();
+    return test_complete_->Wait(kTestTimeOutLimit);
+  }
+
+  virtual void SetUp() {
+    // Read parameters.
+    queue_size_ = static_cast<size_t>(testing::get<0>(GetParam()));
+    max_frame_length_ = static_cast<size_t>(testing::get<1>(GetParam()));
+    data_length_ = static_cast<size_t>(testing::get<2>(GetParam()));
+    data_type_ = static_cast<DataType>(testing::get<3>(GetParam()));
+
+    // llocate read and write buffers.
+    buffer_reader_flt_ = std::vector<float>(max_frame_length_, 0.0f);
+    buffer_writer_flt_ = std::vector<float>(max_frame_length_, 0.0f);
+    buffer_reader_int16_ = std::vector<int16_t>(max_frame_length_, 0);
+    buffer_writer_int16_ = std::vector<int16_t>(max_frame_length_, 0);
+
+    // Create queues.
+    sample_queue_flt_.reset(
+        new SwapQueue<std::vector<float>>(queue_size_, buffer_reader_flt_));
+    sample_queue_int16_.reset(
+        new SwapQueue<std::vector<int16_t>>(queue_size_, buffer_reader_int16_));
+
+    // Create and populate data vectors.
+    CreateTestData(data_length_);
+  }
+
+  virtual void TearDown() {
+    reader_thread_->Stop();
+    writer_thread_->Stop();
+  }
+
+ private:
+  const int kTestTimeOutLimit = 10 * 1000;
+
+  // Populates the data vectors with random values.
+  void CreateTestData(size_t data_length) {
+    samples_int16_.resize(data_length);
+    samples_float_.resize(data_length);
+    for (size_t k = 0; k < data_length; k++) {
+      samples_float_[k] =
+          (static_cast<float>((rand_r(&writer_seed) % (32768 + 32768 + 1)) -
+                              32768) /
+           32768.0f);
+      samples_int16_[k] = ((rand_r(&writer_seed) % (32767 + 32768)) - 32768);
+    }
+  }
+
+  // Thread callback for the reader thread
+  static bool CbReaderThread(void* context) {
+    return reinterpret_cast<SwapQueueSampleTest*>(context)->CbReaderImpl();
+  }
+
+  // Thread callback for the writer thread
+  static bool CbWriterThread(void* context) {
+    return reinterpret_cast<SwapQueueSampleTest*>(context)->CbWriterImpl();
+  }
+
+  // Tests in a threadsafe manner whether all the samples have been
+  // written and read.
+  bool TestDone() { return (GetNumSamplesRead() == data_length_); }
+
+  // Returns in a threadsafe manner the number of frames read from the queue.
+  size_t GetNumFramesRead() {
+    rtc::CritScope cs(&crit_);
+    return num_frames_read_;
+  }
+
+  // Returns in a threadsafe manner the number of samples read from the queue.
+  size_t GetNumSamplesRead() {
+    rtc::CritScope cs(&crit_);
+    return num_samples_read_;
+  }
+
+  // Increases in a threadsafe manner the number of frames and samples
+  // read from the queue.
+  void IncreaseReadCounters(size_t num_samples_read) {
+    rtc::CritScope cs(&crit_);
+    num_frames_read_++;
+    num_samples_read_ += num_samples_read;
+  }
+
+  // Implements the callback functionality for the reader thread.
+  bool CbReaderImpl() {
+    SleepRandomTime(3, &reader_seed);
+
+    // Read the samples and verify bitexactness.
+    size_t num_samples_read = GetNumSamplesRead();
+    if ((data_type_ == DataType::Float) &&
+        (sample_queue_flt_->Remove(&buffer_reader_flt_))) {
+      for (size_t k = 0; k < buffer_reader_flt_.size(); k++)
+        EXPECT_EQ(buffer_reader_flt_[k], samples_float_[num_samples_read + k]);
+
+      IncreaseReadCounters(buffer_reader_flt_.size());
+    } else if ((data_type_ == DataType::Int16) &&
+               (sample_queue_int16_->Remove(&buffer_reader_int16_))) {
+      for (size_t k = 0; k < buffer_reader_int16_.size(); k++)
+        EXPECT_EQ(buffer_reader_int16_[k],
+                  samples_int16_[num_samples_read + k]);
+
+      IncreaseReadCounters(buffer_reader_int16_.size());
+    }
+
+    // End the test if the test is done or an assert has occurred.
+    if (TestDone() || HasFatalFailure()) {
+      test_complete_->Set();
+    }
+
+    return true;
+  }
+
+  // Implements the callback functionality for the writer thread.
+  bool CbWriterImpl() {
+    SleepRandomTime(3, &writer_seed);
+
+    // Choose number of samples to write.
+    const size_t num_samples_to_write =
+        std::min(rand_r(&writer_seed) % max_frame_length_,
+                 data_length_ - num_samples_written_);
+
+    // Write the data.
+    bool data_written = false;
+    if (num_samples_to_write > 0) {
+      if (data_type_ == DataType::Float) {
+        if (buffer_writer_flt_.size() != num_samples_to_write)
+          buffer_writer_flt_.resize(num_samples_to_write);
+        memcpy(&buffer_writer_flt_[0], &samples_float_[num_samples_written_],
+               num_samples_to_write * sizeof(samples_float_[0]));
+
+        data_written = sample_queue_flt_->Insert(&buffer_writer_flt_);
+      } else {
+        if (buffer_writer_int16_.size() != num_samples_to_write)
+          buffer_writer_int16_.resize(num_samples_to_write);
+
+        memcpy(&buffer_writer_int16_[0], &samples_int16_[num_samples_written_],
+               num_samples_to_write * sizeof(samples_int16_[0]));
+
+        data_written = sample_queue_int16_->Insert(&buffer_writer_int16_);
+      }
+    }
+
+    // Update the number of samples left to write
+    if (data_written) {
+      num_samples_written_ += num_samples_to_write;
+      num_frames_written_++;
+    }
+
+    // End the test early if a fatal failure (ASSERT_*) has occurred.
+    if (HasFatalFailure())
+      test_complete_->Set();
+
+    return true;
+  }
+
+  // Sleeps a random time.
+  static void SleepRandomTime(int max_sleep, unsigned int* seed) {
+    SleepMs(rand_r(seed) % (max_sleep + 1));
+  }
+
+  // Start the threads used in the test.
+  void StartThreads() {
+    EXPECT_TRUE(reader_thread_->Start());
+    reader_thread_->SetPriority(kRealtimePriority);
+    EXPECT_TRUE(writer_thread_->Start());
+    writer_thread_->SetPriority(kRealtimePriority);
+  }
+
+  // Test parameters.
+  size_t queue_size_;
+  size_t max_frame_length_;
+  size_t data_length_;
+  DataType data_type_;
+
+  rtc::scoped_ptr<SwapQueue<std::vector<float>>> sample_queue_flt_;
+  rtc::scoped_ptr<SwapQueue<std::vector<int16_t>>> sample_queue_int16_;
+  rtc::CriticalSection crit_;
+  const rtc::scoped_ptr<EventWrapper> test_complete_;
+  rtc::scoped_ptr<ThreadWrapper> reader_thread_;
+  rtc::scoped_ptr<ThreadWrapper> writer_thread_;
+  std::vector<int16_t> samples_int16_;
+  std::vector<float> samples_float_;
+  size_t num_samples_read_ GUARDED_BY(crit_) = 0;
+  size_t num_samples_written_ = 0;
+  unsigned int reader_seed = 42;
+  unsigned int writer_seed = 37;
+  std::vector<float> buffer_reader_flt_;
+  std::vector<float> buffer_writer_flt_;
+  std::vector<int16_t> buffer_reader_int16_;
+  std::vector<int16_t> buffer_writer_int16_;
+  size_t num_frames_written_ = 0;
+  size_t num_frames_read_ GUARDED_BY(crit_) = 0;
+};
+
+}  // anonymous namespace
+
+// Test parameters for the message queue tests.
+const size_t kMessageQueueSizes[] = {2, 7, 20};
+const size_t kMessageQueueDataLengths[] = {100};
+
+// Test parameters for the sample queue tests.
+const size_t kSampleQueueSizes[] = {2, 7, 100};
+const size_t kMaxFrameLengths[] = {77, 160};
+const size_t kSampleQueueDataLengths[] = {1000, 10000};
+const DataType kDataTypes[] = {DataType::Float, DataType::Int16};
+
+TEST_P(SwapQueueSampleTest, BitExactness) {
+  // Run test and verify that it did not time out.
+  EXPECT_EQ(kEventSignaled, RunTest());
+}
+
+TEST_P(SwapQueueMessageTest, BitExactness) {
+  // Run test and verify that it did not time out.
+  EXPECT_EQ(kEventSignaled, RunTest());
+}
+
+INSTANTIATE_TEST_CASE_P(
+    SwapQueueMessageTestAllCombinations,
+    SwapQueueMessageTest,
+    testing::Combine(::testing::ValuesIn(kMessageQueueSizes),
+                     ::testing::ValuesIn(kMessageQueueDataLengths)));
+
+INSTANTIATE_TEST_CASE_P(
+    SwapQueueSampleTestAllCombinations,
+    SwapQueueSampleTest,
+    testing::Combine(::testing::ValuesIn(kSampleQueueSizes),
+                     ::testing::ValuesIn(kMaxFrameLengths),
+                     ::testing::ValuesIn(kSampleQueueDataLengths),
+                     ::testing::ValuesIn(kDataTypes)));
+
+}  // namespace webrtc