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.
+class ThreadSafeRandomNumberGenerator {

[kwiberg-webrtc, 2015/10/13 13:54:24]

It's not thread safe, it just avoids being terribly thread-unsafe by not having
any global mutable state...

[peah-webrtc, 2015/10/14 12:16:48]

Fully true, I rewrote it using a functor instead.

+ public:
+  ThreadSafeRandomNumberGenerator() : ThreadSafeRandomNumberGenerator(10000) {}
+  explicit ThreadSafeRandomNumberGenerator(int size) {
+    srand(42);
+    random_numbers_.resize(size);
+    for (int k = 0; k < size; k++) {
+      random_numbers_[k] = rand();

[the sun, 2015/10/14 13:05:21]

Note that it doesn't matter that we make rand() thread safe here. Some other
thread could be using it anyway.

There's a simple PRNG class in the BWE test framework that we could possibly
move into webrtc/test/ and reuse. It doesn't even need a lock, since then each
thread could have its own instance.

https://code.google.com/p/chromium/codesearch#chromium/src/third_party/webrtc...

[peah-webrtc, 2015/10/15 07:34:59]

You are fully correct! If ok with you, I'll keep the functor implementation and
use rand in that. That removes the warning at least.

[the sun, 2015/10/15 08:44:00]

Not ok. It still has global mutable state, implicitly, via the rand()
implementation.

Tests should be as deterministic as possible. If you're running multiple threads
using rand(), you cannot make sure that the sequence of random numbers for each
thread is the same for successive test runs. Your initial class tried to solve
this by pre-generating a shared array of N PRNs, but fell short on thread safety
when accessing the shared user_indices_ map.

I'd really rather like it if we had a simple PRNG class that can be instantiated
on each thread.

[peah-webrtc, 2015/10/26 08:56:57]

I definitely agree on the issues with non-determinism in tests and I think that
in this case the lock test is not at all motivated as the queue lock is so
simple. I'll remove it.

+    }
+  }
+
+  // Initializes a user of the random number generator and
+  // returns a user id to use when aquiring a random number.
+  int AddUser(int seed) {
+    int next_user_index = GetNextIserId();
+
+    user_indices_[next_user_index] = seed % random_numbers_.size();
+    return next_user_index;
+  }
+
+  // Returns a random number between 0 and RAND_MAX.
+  int Rand(int user_id) {
+    int index = user_indices_[user_id];
+    index = (index < (static_cast<int>(random_numbers_.size()) - 1) ? index + 1
+                                                                    : 0);
+    user_indices_[user_id] = index;
+    return random_numbers_[index];
+  }

[kwiberg-webrtc, 2015/10/13 13:54:24]

Why do you need to support several isers? Couldn't you just give them their own
instances?

[peah-webrtc, 2015/10/14 12:16:48]

Fully true, I rewrote it using a functor instead.

+
+ private:
+  int GetNextIserId() {
+    int max_user_id = -1;
+    for (auto it = user_indices_.begin(); it != user_indices_.end(); ++it) {
+      max_user_id = std::max(max_user_id, it->first);
+    }
+
+    return max_user_id + 1;
+  }
+
+  std::vector<int> random_numbers_;
+  std::map<int, int> user_indices_;
+};

[kwiberg-webrtc, 2015/10/13 13:54:24]

Wouldn't it be simpler to just protect rand() calls with a lock?

[peah-webrtc, 2015/10/14 12:16:48]

Fully true! This is definitely over-engineered! I rewrote it using a functor
instead.

[the sun, 2015/10/15 08:44:00]

See above comment. You will still have non-determinism, just with a little less
probability of pulling the same number from the PRNG hat in several threads
simultaneously.

[peah-webrtc, 2015/10/26 08:56:57]

Acknowledged.

+
+// 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);
+
+    // Setup the random number generator.
+    reader_rand_id_ = random_generator_.AddUser(42);
+    writer_rand_id_ = random_generator_.AddUser(37);
+  }
+
+  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_.Rand(writer_rand_id_) % 10000,
+                               random_generator_.Rand(writer_rand_id_) % 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, reader_rand_id_);
+
+    // 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, writer_rand_id_);
+
+    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, int user_id) {
+    SleepMs(random_generator_.Rand(user_id) % (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_;
+  int reader_rand_id_;
+  int writer_rand_id_;
+  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<std::vector<int16_t>> template_queue(queue_size_);
+    for (size_t k = 0; k < queue_size_; k++) {
+      template_queue.resize(buffer_reader_.size());
+    }
+    sample_queue_.reset(new SwapQueue<std::vector<int16_t>>(&template_queue));
+
+    // Create and populate data vectors.
+    CreateTestData(data_length_);
+
+    // Setup the random number generator.
+    reader_rand_id_ = random_generator_.AddUser(42);
+    writer_rand_id_ = random_generator_.AddUser(37);
+  }
+
+  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_.Rand(writer_rand_id_) % (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_rand_id_);
+
+    // 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, writer_rand_id_);
+
+    // Choose number of samples to write.
+    const size_t num_samples_to_write =
+        std::min(random_generator_.Rand(writer_rand_id_) % 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, int user_id) {
+    SleepMs(random_generator_.Rand(user_id) % (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;
+  int reader_rand_id_;
+  int writer_rand_id_;
+  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 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};
+
+}  // anonymous namespace
+
+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) {
+  const size_t kQueueSize = 3;
+  std::vector<std::vector<int>> v(2);
+  v[0].resize(kQueueSize);
+  v[1].resize(kQueueSize);
+  SwapQueue<std::vector<int>> queue(&v);
+  std::vector<int> i(kQueueSize, 0);
+
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_EQ(i.size(), kQueueSize);
+  EXPECT_TRUE(queue.Insert(&i));
+  EXPECT_EQ(i.size(), kQueueSize);
+  EXPECT_TRUE(queue.Remove(&i));
+  EXPECT_EQ(i.size(), kQueueSize);
+  EXPECT_TRUE(queue.Remove(&i));
+  EXPECT_EQ(i.size(), kQueueSize);
+}
+
+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