Changed queue implementation to the proposed vector-based solution.

webrtc/modules/audio_processing/swapped_nonblocking_queue_unittest.cc

+/*
+ *  Copyright (c) 2014 The WebRTC project authors. All Rights Reserved.

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

You do seem to like multiples of 53...

[peah-webrtc, 2015/10/12 13:01:08]

It is a prime number :-)

But is it occurring in this file as well?

[hlundin-webrtc, 2015/10/13 06:37:58]

I think 53 is just a coincident. 19 is obviously the factor of desire.

[peah-webrtc, 2015/10/13 11:56:12]

:-) It is a shame 42 is no prime number.

+ *
+ *  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/swapped_nonblocking_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 {
+
+// Sample data type enum.
+enum class DataType { Float, Int16 };
+
+// The type for a message.
+class Message {
+ public:
+  Message() {
+    id_ = -1;
+    info_ = -1;
+  }
+
+  explicit Message(const Message& m) {

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

Copy constructor shouldn't be explicit.

[peah-webrtc, 2015/10/12 13:01:08]

Done.

+    id_ = m.id_;
+    info_ = m.info_;
+  }
+
+  explicit Message(int id, int info) {

[hlundin-webrtc, 2015/10/13 06:37:57]

No need to make a 2-argument constructor explicit, is it?

[peah-webrtc, 2015/10/13 11:56:12]

True! Removed that.

+    id_ = id;
+    info_ = info;
+  }
+
+  bool Compare(const Message& m) {
+    return ((id_ == m.id_) && (info_ == m.info_));
+  }
+
+  void swap(Message& m) {
+    int tmp = id_;
+    id_ = m.id_;
+    m.id_ = tmp;
+
+    tmp = info_;
+    info_ = m.info_;
+    m.info_ = tmp;

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

std::swap(id_, m.id_);
std::swap(info_, m.info_);

[peah-webrtc, 2015/10/12 13:01:08]

Done.

+  }
+
+ private:
+  int id_;
+  int info_;
+};
+
+// Implements the tests for the sample queue.
+class SwappedNonBlockingQueueMessageTest
+    : public ::testing::TestWithParam<::testing::tuple<size_t, size_t>> {
+ public:
+  SwappedNonBlockingQueueMessageTest()
+      : 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 SwappedNonBlockingQueue<Message>(queue_size));
+
+    // Create and populate message vector.
+    messages_ = std::vector<Message>();
+    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] =
+          Message(rand_r(&writer_seed) % 10000, rand_r(&writer_seed) % 10000);
+  }
+
+  // Thread callback for the reader thread

[hlundin-webrtc, 2015/10/13 06:37:57]

End with .

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+  static bool CbReaderThread(void* context) {
+    return reinterpret_cast<SwappedNonBlockingQueueMessageTest*>(context)
+        ->CbReaderImpl();
+  }
+
+  // Thread callback for the writer thread

[hlundin-webrtc, 2015/10/13 06:37:57]

End with .

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+  static bool CbWriterThread(void* context) {
+    return reinterpret_cast<SwappedNonBlockingQueueMessageTest*>(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() {

[hlundin-webrtc, 2015/10/13 06:37:57]

const-declare the method (and make crit_ mutable).

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+    rtc::CritScope cs(&crit_);
+    return num_messages_read_;

[hlundin-webrtc, 2015/10/13 06:37:57]

I'm not sure, but this seems like the kind of thing that should be able to
achieve with an atomic instead of using locks. One thread is making simple
increments, and the other is reading the value.

[peah-webrtc, 2015/10/13 11:56:11]

That is fully true. But I think for the purpose of the test it does not matter.
I've minimized the time the lock is held.

+  }
+
+  // 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();

[hlundin-webrtc, 2015/10/13 06:37:58]

const
But, you don't need this variable, at least not here. Either move it inside the
if-statement below, or simply use the method invocation as index:
messages_[GetNumMessagesRead()]

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+    Message 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.
+      Message 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));

[hlundin-webrtc, 2015/10/13 06:37:57]

Will rand_r work across platforms?

[peah-webrtc, 2015/10/13 11:56:12]

Great find!
I implemented a threadsafe random generator instead to both get platform
compatibility and void lint warnings about threadsafety of rand().

+  }
+
+  // Start the threads used in the test.
+  void StartThreads() {
+    EXPECT_TRUE(reader_thread_->Start());

[hlundin-webrtc, 2015/10/13 06:37:57]

ASSERT_TRUE. Makes little sense to continue if thread-starting fails.

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+    reader_thread_->SetPriority(kRealtimePriority);
+    EXPECT_TRUE(writer_thread_->Start());
+    writer_thread_->SetPriority(kRealtimePriority);
+  }
+
+  // Test parameters.
+  rtc::scoped_ptr<SwappedNonBlockingQueue<Message>> 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<Message> 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 SwappedNonBlockingQueueSampleTest
+    : public ::testing::TestWithParam<
+          ::testing::tuple<size_t, size_t, size_t, DataType>> {
+ public:
+  SwappedNonBlockingQueueSampleTest()
+      : 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()));

[hlundin-webrtc, 2015/10/13 06:37:57]

If you parse these parameters in the constructor instead of in SetUp, you can
make all these member variables const.

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+    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.

[hlundin-webrtc, 2015/10/13 06:37:57]

-> Allocate

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+    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 SwappedNonBlockingQueue<std::vector<float>>(
+        queue_size_, buffer_reader_flt_));
+    sample_queue_int16_.reset(new SwappedNonBlockingQueue<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

[hlundin-webrtc, 2015/10/13 06:37:57]

End with .

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+  static bool CbReaderThread(void* context) {
+    return reinterpret_cast<SwappedNonBlockingQueueSampleTest*>(context)
+        ->CbReaderImpl();
+  }
+
+  // Thread callback for the writer thread

[hlundin-webrtc, 2015/10/13 06:37:57]

End with .

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+  static bool CbWriterThread(void* context) {
+    return reinterpret_cast<SwappedNonBlockingQueueSampleTest*>(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();

[hlundin-webrtc, 2015/10/13 06:37:57]

const

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+    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());

[hlundin-webrtc, 2015/10/13 06:37:57]

Again, ASSERT_*?

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+    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<SwappedNonBlockingQueue<std::vector<float>>>
+      sample_queue_flt_;
+  rtc::scoped_ptr<SwappedNonBlockingQueue<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;

[hlundin-webrtc, 2015/10/13 06:37:57]

Trailing underscore, please.

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+  unsigned int writer_seed = 37;

[hlundin-webrtc, 2015/10/13 06:37:58]

Trailing underscore, please.

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+  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.

[hlundin-webrtc, 2015/10/13 06:37:57]

Let the consts live in the anonymous namespace too.

[peah-webrtc, 2015/10/13 11:56:12]

Done.

+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(SwappedNonBlockingQueueSampleTest, BitExactness) {
+  // Run test and verify that it did not time out.
+  EXPECT_EQ(kEventSignaled, RunTest());
+}
+
+TEST_P(SwappedNonBlockingQueueMessageTest, BitExactness) {
+  // Run test and verify that it did not time out.
+  EXPECT_EQ(kEventSignaled, RunTest());
+}
+
+INSTANTIATE_TEST_CASE_P(
+    SwappedNonBlockingQueueMessageTestAllCombinations,
+    SwappedNonBlockingQueueMessageTest,
+    testing::Combine(::testing::ValuesIn(kMessageQueueSizes),
+                     ::testing::ValuesIn(kMessageQueueDataLengths)));
+
+INSTANTIATE_TEST_CASE_P(
+    SwappedNonBlockingQueueSampleTestAllCombinations,
+    SwappedNonBlockingQueueSampleTest,
+    testing::Combine(::testing::ValuesIn(kSampleQueueSizes),
+                     ::testing::ValuesIn(kMaxFrameLengths),
+                     ::testing::ValuesIn(kSampleQueueDataLengths),
+                     ::testing::ValuesIn(kDataTypes)));
+
+}  // namespace webrtc

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

It might just be me, but I think testing of the queue should focus less on
thread safety (which is almost impossible to test comprehensively anyway) since
that's best established by code review and the annotations you've put in,
especially for a piece of code this small and simple. And more on correctness
issues such as verifying that Insert fails iff the queue is already full, corner
cases such as size == 1, checking that the queue will never do anything to the
Ts except swap them, and so on.

[peah-webrtc, 2015/10/13 11:56:12]

I agree, and I add some tests for the corner cases, but keep the existing tests.
The current test configuration take 1 second to run.