Changed queue implementation to the proposed vector-based solution.

webrtc/common_audio/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 <map>
+#include <tuple>
+#include <vector>
+
+#include "testing/gtest/include/gtest/gtest.h"
+#include "webrtc/common_audio/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 {
+
+// Provides thread_safe random numbers.

[kwiberg-webrtc, 2015/10/14 14:10:16]

The comment says nothing that the class name doesn't already say.

[peah-webrtc, 2015/10/15 07:35:00]

Good point! I removed the comment.

[peah-webrtc, 2015/10/15 07:35:00]

Done.

+class ThreadSafeRandomNumberGenerator {
+ public:
+  ThreadSafeRandomNumberGenerator() : ThreadSafeRandomNumberGenerator(42) {}
+
+  explicit ThreadSafeRandomNumberGenerator(int seed) { srand(seed); }
+
+  int operator()() {
+    rtc::CritScope cs(&crit_);
+    return rand();
+  }
+
+ private:
+  rtc::CriticalSection crit_;
+};
+
+// The type for a message.
+class SwapQueueTestMessage {
+ public:
+  SwapQueueTestMessage() {
+    id_ = -1;
+    info_ = -1;
+  }
+
+  SwapQueueTestMessage(const SwapQueueTestMessage& m)
+      : id_(m.id_), info_(m.info_) {}
+
+  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_;
+};
+
+// 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(random_generator_() % 10000,
+                                          random_generator_() % 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() const {
+    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);
+
+    // Read the message and verify bitexactness.
+    SwapQueueTestMessage m;
+    if (queue_->Remove(&m)) {
+      EXPECT_TRUE(m.Compare(messages_[GetNumMessagesRead()]));
+      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);
+
+    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.
+  void SleepRandomTime(int max_sleep) {
+    SleepMs(random_generator_() % (max_sleep + 1));
+  }
+
+  // Start the threads used in the test.
+  void StartThreads() {
+    ASSERT_TRUE(reader_thread_->Start());
+    reader_thread_->SetPriority(kRealtimePriority);
+    ASSERT_TRUE(writer_thread_->Start());
+    writer_thread_->SetPriority(kRealtimePriority);
+  }
+
+  // Test parameters.
+  rtc::scoped_ptr<SwapQueue<SwapQueueTestMessage>> queue_;
+  mutable rtc::CriticalSection crit_;
+  ThreadSafeRandomNumberGenerator random_generator_;
+  const rtc::scoped_ptr<EventWrapper> test_complete_;
+  rtc::scoped_ptr<ThreadWrapper> reader_thread_;
+  rtc::scoped_ptr<ThreadWrapper> writer_thread_;
+  std::vector<SwapQueueTestMessage> messages_;
+  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>> {
+ public:
+  SwapQueueSampleTest()
+      : 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()))),
+        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() {
+    // Allocate read and write buffers.
+    buffer_reader_ = std::vector<int16_t>(max_frame_length_, 0);
+    buffer_writer_ = std::vector<int16_t>(max_frame_length_, 0);
+
+    // Create queue.
+    std::vector<int16_t> template_queue_element(buffer_reader_.size());
+    sample_queue_.reset(new SwapQueue<std::vector<int16_t>>(
+        queue_size_, template_queue_element));
+
+    // 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_.resize(data_length);
+    for (size_t k = 0; k < data_length; k++) {
+      samples_[k] = ((random_generator_() % (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);
+
+    // Read the samples and verify bitexactness.
+    const size_t num_samples_read = GetNumSamplesRead();
+    if (sample_queue_->Remove(&buffer_reader_)) {
+      for (size_t k = 0; k < buffer_reader_.size(); k++)
+        EXPECT_EQ(buffer_reader_[k], samples_[num_samples_read + k]);
+
+      IncreaseReadCounters(buffer_reader_.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);
+
+    // Choose number of samples to write.
+    const size_t num_samples_to_write =
+        std::min(random_generator_() % max_frame_length_,
+                 data_length_ - num_samples_written_);
+
+    // Write the data.
+    bool data_written = false;
+    if (num_samples_to_write > 0) {
+      if (buffer_writer_.size() != num_samples_to_write)
+        buffer_writer_.resize(num_samples_to_write);
+
+      memcpy(&buffer_writer_[0], &samples_[num_samples_written_],
+             num_samples_to_write * sizeof(samples_[0]));
+
+      data_written = sample_queue_->Insert(&buffer_writer_);
+    }
+
+    // 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.
+  void SleepRandomTime(int max_sleep) {
+    SleepMs(random_generator_() % (max_sleep + 1));
+  }
+
+  // Start the threads used in the test.
+  void StartThreads() {
+    ASSERT_TRUE(reader_thread_->Start());
+    reader_thread_->SetPriority(kRealtimePriority);
+    ASSERT_TRUE(writer_thread_->Start());
+    writer_thread_->SetPriority(kRealtimePriority);
+  }
+
+  // Test parameters.
+  const size_t queue_size_;
+  const size_t max_frame_length_;
+  const size_t data_length_;
+
+  rtc::scoped_ptr<SwapQueue<std::vector<int16_t>>> sample_queue_;
+  rtc::CriticalSection crit_;
+  ThreadSafeRandomNumberGenerator random_generator_;
+  const rtc::scoped_ptr<EventWrapper> test_complete_;
+  rtc::scoped_ptr<ThreadWrapper> reader_thread_;
+  rtc::scoped_ptr<ThreadWrapper> writer_thread_;
+  std::vector<int16_t> samples_;
+  size_t num_samples_read_ GUARDED_BY(crit_) = 0;
+  size_t num_samples_written_ = 0;
+  std::vector<int16_t> buffer_reader_;
+  std::vector<int16_t> buffer_writer_;
+  size_t num_frames_written_ = 0;
+  size_t num_frames_read_ GUARDED_BY(crit_) = 0;
+};
+
+// Test parameter for the basic sample based SwapQueue Tests.
+const size_t kChunkSize = 3;
+
+// 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[] = {7, 100};
+const size_t kMaxFrameLengths[] = {77, 160};
+const size_t kSampleQueueDataLengths[] = {200, 500};
+
+// Item invariance check function for the InvarianceVerification
+// test.
+bool InvarianceVerifier(const std::vector<int>& v) {
+  RTC_CHECK(v.size() == kChunkSize);
+  return true;
+}
+
+}  // anonymous namespace
+
+TEST(SwapQueueTest, SuccessfulInvarianceVerification) {
+  std::vector<int> template_element(kChunkSize);
+  SwapQueue<std::vector<int>, &InvarianceVerifier> queue(2, template_element);
+  std::vector<int> valid_chunk(kChunkSize, 0);
+
+  EXPECT_TRUE(queue.Insert(&valid_chunk));
+  EXPECT_EQ(valid_chunk.size(), kChunkSize);
+  EXPECT_TRUE(queue.Remove(&valid_chunk));
+  EXPECT_EQ(valid_chunk.size(), kChunkSize);
+}
+
+#if defined(GTEST_HAS_DEATH_TEST)
+TEST(SwapQueueTest, UnSuccessfulInvarianceVerification1) {
+  std::vector<int> template_element(kChunkSize);
+  SwapQueue<std::vector<int>, &InvarianceVerifier> queue(2, template_element);
+  std::vector<int> invalid_chunk(kChunkSize - 1, 0);
+  EXPECT_DEATH(queue.Insert(&invalid_chunk), "");
+  EXPECT_DEATH(queue.Insert(&invalid_chunk), "");

[kwiberg-webrtc, 2015/10/14 14:10:16]

Unless your employee ID is 007, you don't get to live twice.

[peah-webrtc, 2015/10/15 07:35:00]

:-)

[peah-webrtc, 2015/10/15 07:35:00]

Done.

+}
+
+TEST(SwapQueueTest, UnSuccessfulInvarianceVerification2) {
+  std::vector<int> template_element(kChunkSize);
+  SwapQueue<std::vector<int>, &InvarianceVerifier> queue(2, template_element);
+  std::vector<int> invalid_chunk(kChunkSize - 1, 0);
+  std::vector<int> valid_chunk(kChunkSize, 0);
+
+  EXPECT_TRUE(queue.Insert(&valid_chunk));
+  EXPECT_EQ(valid_chunk.size(), kChunkSize);
+  EXPECT_DEATH(queue.Remove(&invalid_chunk), "");
+}
+#endif
+
+TEST(SwapQueueTest, FullQueue) {
+  SwapQueue<int> queue(2);
+  int i = 0;
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_FALSE(queue.Insert(&i));
+  EXPECT_TRUE(queue.Remove(&i));
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_FALSE(queue.Insert(&i));
+}
+
+TEST(SwapQueueTest, EmptyQueue) {
+  SwapQueue<int> queue(2);
+  int i = 0;
+  EXPECT_FALSE(queue.Remove(&i));
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_TRUE(queue.Remove(&i));
+  EXPECT_FALSE(queue.Remove(&i));
+}
+
+TEST(SwapQueueTest, InitializeTest) {
+  std::vector<int> i(kChunkSize, 0);
+  SwapQueue<std::vector<int>> queue(2, i);
+
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_EQ(i.size(), kChunkSize);
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_EQ(i.size(), kChunkSize);
+  EXPECT_TRUE(queue.Remove(&i));
+  EXPECT_EQ(i.size(), kChunkSize);
+  EXPECT_TRUE(queue.Remove(&i));
+  EXPECT_EQ(i.size(), kChunkSize);
+}
+
+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)));
+
+}  // namespace webrtc