From 48407f71221635aec2d2a382ee24dea28e4f69cf Mon Sep 17 00:00:00 2001 From: peah Date: Mon, 9 Nov 2015 05:24:50 -0800 Subject: [PATCH] Changed queue implementation to the proposed vector-based solution. Added unit tests. BUG=webrtc:5099 TBR=hlundin-webrtc Review URL: https://codereview.webrtc.org/1398473004 Cr-Commit-Position: refs/heads/master@{#10562} --- webrtc/common_audio/BUILD.gn | 1 + webrtc/common_audio/common_audio.gyp | 2 + webrtc/common_audio/swap_queue.h | 210 +++++++++++++++++++ webrtc/common_audio/swap_queue_unittest.cc | 225 +++++++++++++++++++++ 4 files changed, 438 insertions(+) create mode 100644 webrtc/common_audio/swap_queue.h create mode 100644 webrtc/common_audio/swap_queue_unittest.cc diff --git a/webrtc/common_audio/BUILD.gn b/webrtc/common_audio/BUILD.gn index b01b31816b..b4ec1d71ef 100644 --- a/webrtc/common_audio/BUILD.gn +++ b/webrtc/common_audio/BUILD.gn @@ -87,6 +87,7 @@ source_set("common_audio") { "signal_processing/vector_scaling_operations.c", "sparse_fir_filter.cc", "sparse_fir_filter.h", + "swap_queue.h", "vad/include/vad.h", "vad/include/webrtc_vad.h", "vad/vad.cc", diff --git a/webrtc/common_audio/common_audio.gyp b/webrtc/common_audio/common_audio.gyp index 884a8afcf8..43c7ac52c5 100644 --- a/webrtc/common_audio/common_audio.gyp +++ b/webrtc/common_audio/common_audio.gyp @@ -101,6 +101,7 @@ 'signal_processing/vector_scaling_operations.c', 'sparse_fir_filter.cc', 'sparse_fir_filter.h', + 'swap_queue.h', 'vad/include/vad.h', 'vad/include/webrtc_vad.h', 'vad/vad.cc', @@ -256,6 +257,7 @@ 'signal_processing/real_fft_unittest.cc', 'signal_processing/signal_processing_unittest.cc', 'sparse_fir_filter_unittest.cc', + 'swap_queue_unittest.cc', 'vad/vad_core_unittest.cc', 'vad/vad_filterbank_unittest.cc', 'vad/vad_gmm_unittest.cc', diff --git a/webrtc/common_audio/swap_queue.h b/webrtc/common_audio/swap_queue.h new file mode 100644 index 0000000000..d8bb5c024e --- /dev/null +++ b/webrtc/common_audio/swap_queue.h @@ -0,0 +1,210 @@ +/* + * 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. + */ + +#ifndef WEBRTC_COMMON_AUDIO_SWAP_QUEUE_H_ +#define WEBRTC_COMMON_AUDIO_SWAP_QUEUE_H_ + +#include +#include +#include + +#include "webrtc/base/checks.h" +#include "webrtc/base/criticalsection.h" + +namespace webrtc { + +namespace internal { + +// (Internal; please don't use outside this file.) +template +bool NoopSwapQueueItemVerifierFunction(const T&) { + return true; +} + +} // namespace internal + +// Functor to use when supplying a verifier function for the queue. +template +class SwapQueueItemVerifier { + public: + bool operator()(const T& t) const { return QueueItemVerifierFunction(t); } +}; + +// This class is a fixed-size queue. A producer calls Insert() to insert +// an element of type T at the back of the queue, and a consumer calls +// Remove() to remove an element from the front of the queue. It's safe +// for the producer(s) and the consumer(s) to access the queue +// concurrently, from different threads. +// +// To avoid the construction, copying, and destruction of Ts that a naive +// queue implementation would require, for each "full" T passed from +// producer to consumer, SwapQueue passes an "empty" T in the other +// direction (an "empty" T is one that contains nothing of value for the +// consumer). This bidirectional movement is implemented with swap(). +// +// // Create queue: +// Bottle proto(568); // Prepare an empty Bottle. Heap allocates space for +// // 568 ml. +// SwapQueue q(N, proto); // Init queue with N copies of proto. +// // Each copy allocates on the heap. +// // Producer pseudo-code: +// Bottle b(568); // Prepare an empty Bottle. Heap allocates space for 568 ml. +// loop { +// b.Fill(amount); // Where amount <= 568 ml. +// q.Insert(&b); // Swap our full Bottle for an empty one from q. +// } +// +// // Consumer pseudo-code: +// Bottle b(568); // Prepare an empty Bottle. Heap allocates space for 568 ml. +// loop { +// q.Remove(&b); // Swap our empty Bottle for the next-in-line full Bottle. +// Drink(&b); +// } +// +// For a well-behaved Bottle class, there are no allocations in the +// producer, since it just fills an empty Bottle that's already large +// enough; no deallocations in the consumer, since it returns each empty +// Bottle to the queue after having drunk it; and no copies along the +// way, since the queue uses swap() everywhere to move full Bottles in +// one direction and empty ones in the other. +template > +class SwapQueue { + public: + // Creates a queue of size size and fills it with default constructed Ts. + explicit SwapQueue(size_t size) : queue_(size) { + RTC_DCHECK(VerifyQueueSlots()); + } + + // Same as above and accepts an item verification functor. + SwapQueue(size_t size, const QueueItemVerifier& queue_item_verifier) + : queue_item_verifier_(queue_item_verifier), queue_(size) { + RTC_DCHECK(VerifyQueueSlots()); + } + + // Creates a queue of size size and fills it with copies of prototype. + SwapQueue(size_t size, const T& prototype) : queue_(size, prototype) { + RTC_DCHECK(VerifyQueueSlots()); + } + + // Same as above and accepts an item verification functor. + SwapQueue(size_t size, + const T& prototype, + const QueueItemVerifier& queue_item_verifier) + : queue_item_verifier_(queue_item_verifier), queue_(size, prototype) { + RTC_DCHECK(VerifyQueueSlots()); + } + + // Resets the queue to have zero content wile maintaining the queue size. + void Clear() { + rtc::CritScope cs(&crit_queue_); + next_write_index_ = 0; + next_read_index_ = 0; + num_elements_ = 0; + } + + // Inserts a "full" T at the back of the queue by swapping *input with an + // "empty" T from the queue. + // Returns true if the item was inserted or false if not (the queue was full). + // When specified, the T given in *input must pass the ItemVerifier() test. + // The contents of *input after the call are then also guaranteed to pass the + // ItemVerifier() test. + bool Insert(T* input) WARN_UNUSED_RESULT { + RTC_DCHECK(input); + + rtc::CritScope cs(&crit_queue_); + + RTC_DCHECK(queue_item_verifier_(*input)); + + if (num_elements_ == queue_.size()) { + return false; + } + + using std::swap; + swap(*input, queue_[next_write_index_]); + + ++next_write_index_; + if (next_write_index_ == queue_.size()) { + next_write_index_ = 0; + } + + ++num_elements_; + + RTC_DCHECK_LT(next_write_index_, queue_.size()); + RTC_DCHECK_LE(num_elements_, queue_.size()); + + return true; + } + + // Removes the frontmost "full" T from the queue by swapping it with + // the "empty" T in *output. + // Returns true if an item could be removed or false if not (the queue was + // empty). When specified, The T given in *output must pass the ItemVerifier() + // test and the contents of *output after the call are then also guaranteed to + // pass the ItemVerifier() test. + bool Remove(T* output) WARN_UNUSED_RESULT { + RTC_DCHECK(output); + + rtc::CritScope cs(&crit_queue_); + + RTC_DCHECK(queue_item_verifier_(*output)); + + if (num_elements_ == 0) { + return false; + } + + using std::swap; + swap(*output, queue_[next_read_index_]); + + ++next_read_index_; + if (next_read_index_ == queue_.size()) { + next_read_index_ = 0; + } + + --num_elements_; + + RTC_DCHECK_LT(next_read_index_, queue_.size()); + RTC_DCHECK_LE(num_elements_, queue_.size()); + + return true; + } + + private: + // Verify that the queue slots complies with the ItemVerifier test. + bool VerifyQueueSlots() { + rtc::CritScope cs(&crit_queue_); + for (const auto& v : queue_) { + RTC_DCHECK(queue_item_verifier_(v)); + } + return true; + } + + rtc::CriticalSection crit_queue_; + + // TODO(peah): Change this to use std::function() once we can use C++11 std + // lib. + QueueItemVerifier queue_item_verifier_ GUARDED_BY(crit_queue_); + + // (next_read_index_ + num_elements_) % queue_.size() = + // next_write_index_ + size_t next_write_index_ GUARDED_BY(crit_queue_) = 0; + size_t next_read_index_ GUARDED_BY(crit_queue_) = 0; + size_t num_elements_ GUARDED_BY(crit_queue_) = 0; + + // queue_.size() is constant. + std::vector queue_ GUARDED_BY(crit_queue_); + + RTC_DISALLOW_COPY_AND_ASSIGN(SwapQueue); +}; + +} // namespace webrtc + +#endif // WEBRTC_COMMON_AUDIO_SWAP_QUEUE_H_ diff --git a/webrtc/common_audio/swap_queue_unittest.cc b/webrtc/common_audio/swap_queue_unittest.cc new file mode 100644 index 0000000000..104e494bc6 --- /dev/null +++ b/webrtc/common_audio/swap_queue_unittest.cc @@ -0,0 +1,225 @@ +/* + * 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 "webrtc/common_audio/swap_queue.h" + +#include + +#include "testing/gtest/include/gtest/gtest.h" + +namespace webrtc { + +namespace { + +// Test parameter for the basic sample based SwapQueue Tests. +const size_t kChunkSize = 3; + +// Queue item verification function for the vector test. +bool LengthVerifierFunction(const std::vector& v) { + return v.size() == kChunkSize; +} + +// Queue item verifier for the vector test. +class LengthVerifierFunctor { + public: + explicit LengthVerifierFunctor(size_t length) : length_(length) {} + + bool operator()(const std::vector& v) const { + return v.size() == length_; + } + + private: + size_t length_; +}; + +} // anonymous namespace + +TEST(SwapQueueTest, BasicOperation) { + std::vector i(kChunkSize, 0); + SwapQueue> 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(SwapQueueTest, FullQueue) { + SwapQueue queue(2); + + // Fill the queue. + int i = 0; + EXPECT_TRUE(queue.Insert(&i)); + i = 1; + EXPECT_TRUE(queue.Insert(&i)); + + // Ensure that the value is not swapped when doing an Insert + // on a full queue. + i = 2; + EXPECT_FALSE(queue.Insert(&i)); + EXPECT_EQ(i, 2); + + // Ensure that the Insert didn't overwrite anything in the queue. + EXPECT_TRUE(queue.Remove(&i)); + EXPECT_EQ(i, 0); + EXPECT_TRUE(queue.Remove(&i)); + EXPECT_EQ(i, 1); +} + +TEST(SwapQueueTest, EmptyQueue) { + SwapQueue 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, Clear) { + SwapQueue queue(2); + int i = 0; + + // Fill the queue. + EXPECT_TRUE(queue.Insert(&i)); + EXPECT_TRUE(queue.Insert(&i)); + + // Ensure full queue. + EXPECT_FALSE(queue.Insert(&i)); + + // Empty the queue. + queue.Clear(); + + // Ensure that the queue is empty + EXPECT_FALSE(queue.Remove(&i)); + + // Ensure that the queue is no longer full. + EXPECT_TRUE(queue.Insert(&i)); +} + +TEST(SwapQueueTest, SuccessfulItemVerifyFunction) { + std::vector template_element(kChunkSize); + SwapQueue, + SwapQueueItemVerifier, LengthVerifierFunction>> + queue(2, template_element); + std::vector 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); +} + +TEST(SwapQueueTest, SuccessfulItemVerifyFunctor) { + std::vector template_element(kChunkSize); + LengthVerifierFunctor verifier(kChunkSize); + SwapQueue, LengthVerifierFunctor> queue(2, template_element, + verifier); + std::vector 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 RTC_DCHECK_IS_ON && GTEST_HAS_DEATH_TEST && !defined(WEBRTC_ANDROID) +TEST(SwapQueueTest, UnsuccessfulItemVerifyFunctor) { + // Queue item verifier for the test. + auto minus_2_verifier = [](const int& i) { return i > -2; }; + SwapQueue queue(2, minus_2_verifier); + + int valid_value = 1; + int invalid_value = -4; + EXPECT_TRUE(queue.Insert(&valid_value)); + EXPECT_TRUE(queue.Remove(&valid_value)); + bool result; + EXPECT_DEATH(result = queue.Insert(&invalid_value), ""); +} + +TEST(SwapQueueTest, UnSuccessfulItemVerifyInsert) { + std::vector template_element(kChunkSize); + SwapQueue, + SwapQueueItemVerifier, &LengthVerifierFunction>> + queue(2, template_element); + std::vector invalid_chunk(kChunkSize - 1, 0); + bool result; + EXPECT_DEATH(result = queue.Insert(&invalid_chunk), ""); +} + +TEST(SwapQueueTest, UnSuccessfulItemVerifyRemove) { + std::vector template_element(kChunkSize); + SwapQueue, + SwapQueueItemVerifier, &LengthVerifierFunction>> + queue(2, template_element); + std::vector invalid_chunk(kChunkSize - 1, 0); + std::vector valid_chunk(kChunkSize, 0); + EXPECT_TRUE(queue.Insert(&valid_chunk)); + EXPECT_EQ(valid_chunk.size(), kChunkSize); + bool result; + EXPECT_DEATH(result = queue.Remove(&invalid_chunk), ""); +} +#endif + +TEST(SwapQueueTest, VectorContentTest) { + const size_t kQueueSize = 10; + const size_t kFrameLength = 160; + const size_t kDataLength = kQueueSize * kFrameLength; + std::vector buffer_reader(kFrameLength, 0); + std::vector buffer_writer(kFrameLength, 0); + SwapQueue> queue(kQueueSize, + std::vector(kFrameLength)); + std::vector samples(kDataLength); + + for (size_t k = 0; k < kDataLength; k++) { + samples[k] = k % 9; + } + + for (size_t k = 0; k < kQueueSize; k++) { + buffer_writer.clear(); + buffer_writer.insert(buffer_writer.end(), &samples[0] + k * kFrameLength, + &samples[0] + (k + 1) * kFrameLength); + + EXPECT_TRUE(queue.Insert(&buffer_writer)); + } + + for (size_t k = 0; k < kQueueSize; k++) { + EXPECT_TRUE(queue.Remove(&buffer_reader)); + + for (size_t j = 0; j < buffer_reader.size(); j++) { + EXPECT_EQ(buffer_reader[j], samples[k * kFrameLength + j]); + } + } +} + +TEST(SwapQueueTest, ZeroSlotQueue) { + SwapQueue queue(0); + int i = 42; + EXPECT_FALSE(queue.Insert(&i)); + EXPECT_FALSE(queue.Remove(&i)); + EXPECT_EQ(i, 42); +} + +TEST(SwapQueueTest, OneSlotQueue) { + SwapQueue queue(1); + int i = 42; + EXPECT_TRUE(queue.Insert(&i)); + i = 43; + EXPECT_FALSE(queue.Insert(&i)); + EXPECT_EQ(i, 43); + EXPECT_TRUE(queue.Remove(&i)); + EXPECT_EQ(i, 42); + EXPECT_FALSE(queue.Remove(&i)); +} + +} // namespace webrtc