250 lines
9.1 KiB
C
250 lines
9.1 KiB
C
|
/*
|
||
|
|
* 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 RTC_BASE_SWAP_QUEUE_H_
|
||
|
|
#define RTC_BASE_SWAP_QUEUE_H_
|
||
|
|
|
||
|
|
#include <stddef.h>
|
||
|
|
|
||
|
|
#include <atomic>
|
||
|
|
#include <utility>
|
||
|
|
#include <vector>
|
||
|
|
|
||
|
|
#include "absl/base/attributes.h"
|
||
|
|
#include "rtc_base/checks.h"
|
||
|
|
|
||
|
|
namespace webrtc {
|
||
|
|
|
||
|
|
namespace internal {
|
||
|
|
|
||
|
|
// (Internal; please don't use outside this file.)
|
||
|
|
template <typename T>
|
||
|
|
bool NoopSwapQueueItemVerifierFunction(const T&) {
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
|
||
|
|
} // namespace internal
|
||
|
|
|
||
|
|
// Functor to use when supplying a verifier function for the queue.
|
||
|
|
template <typename T,
|
||
|
|
bool (*QueueItemVerifierFunction)(const T&) =
|
||
|
|
internal::NoopSwapQueueItemVerifierFunction>
|
||
|
|
class SwapQueueItemVerifier {
|
||
|
|
public:
|
||
|
|
bool operator()(const T& t) const { return QueueItemVerifierFunction(t); }
|
||
|
|
};
|
||
|
|
|
||
|
|
// This class is a fixed-size queue. A single producer calls Insert() to insert
|
||
|
|
// an element of type T at the back of the queue, and a single consumer calls
|
||
|
|
// Remove() to remove an element from the front of the queue. It's safe for the
|
||
|
|
// producer and the consumer 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<T> 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<Bottle> 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 <typename T, typename QueueItemVerifier = SwapQueueItemVerifier<T>>
|
||
|
|
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 while maintaining the queue size.
|
||
|
|
// Just like Remove(), this can only be called (safely) from the
|
||
|
|
// consumer.
|
||
|
|
void Clear() {
|
||
|
|
// Drop all non-empty elements by resetting num_elements_ and incrementing
|
||
|
|
// next_read_index_ by the previous value of num_elements_. Relaxed memory
|
||
|
|
// ordering is sufficient since the dropped elements are not accessed.
|
||
|
|
next_read_index_ += std::atomic_exchange_explicit(
|
||
|
|
&num_elements_, size_t{0}, std::memory_order_relaxed);
|
||
|
|
if (next_read_index_ >= queue_.size()) {
|
||
|
|
next_read_index_ -= queue_.size();
|
||
|
|
}
|
||
|
|
|
||
|
|
RTC_DCHECK_LT(next_read_index_, queue_.size());
|
||
|
|
}
|
||
|
|
|
||
|
|
// 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.
|
||
|
|
ABSL_MUST_USE_RESULT bool Insert(T* input) {
|
||
|
|
RTC_DCHECK(input);
|
||
|
|
|
||
|
|
RTC_DCHECK(queue_item_verifier_(*input));
|
||
|
|
|
||
|
|
// Load the value of num_elements_. Acquire memory ordering prevents reads
|
||
|
|
// and writes to queue_[next_write_index_] to be reordered to before the
|
||
|
|
// load. (That element might be accessed by a concurrent call to Remove()
|
||
|
|
// until the load finishes.)
|
||
|
|
if (std::atomic_load_explicit(&num_elements_, std::memory_order_acquire) ==
|
||
|
|
queue_.size()) {
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
|
||
|
|
using std::swap;
|
||
|
|
swap(*input, queue_[next_write_index_]);
|
||
|
|
|
||
|
|
// Increment the value of num_elements_ to account for the inserted element.
|
||
|
|
// Release memory ordering prevents the reads and writes to
|
||
|
|
// queue_[next_write_index_] to be reordered to after the increment. (Once
|
||
|
|
// the increment has finished, Remove() might start accessing that element.)
|
||
|
|
const size_t old_num_elements = std::atomic_fetch_add_explicit(
|
||
|
|
&num_elements_, size_t{1}, std::memory_order_release);
|
||
|
|
|
||
|
|
++next_write_index_;
|
||
|
|
if (next_write_index_ == queue_.size()) {
|
||
|
|
next_write_index_ = 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
RTC_DCHECK_LT(next_write_index_, queue_.size());
|
||
|
|
RTC_DCHECK_LT(old_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.
|
||
|
|
ABSL_MUST_USE_RESULT bool Remove(T* output) {
|
||
|
|
RTC_DCHECK(output);
|
||
|
|
|
||
|
|
RTC_DCHECK(queue_item_verifier_(*output));
|
||
|
|
|
||
|
|
// Load the value of num_elements_. Acquire memory ordering prevents reads
|
||
|
|
// and writes to queue_[next_read_index_] to be reordered to before the
|
||
|
|
// load. (That element might be accessed by a concurrent call to Insert()
|
||
|
|
// until the load finishes.)
|
||
|
|
if (std::atomic_load_explicit(&num_elements_, std::memory_order_acquire) ==
|
||
|
|
0) {
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
|
||
|
|
using std::swap;
|
||
|
|
swap(*output, queue_[next_read_index_]);
|
||
|
|
|
||
|
|
// Decrement the value of num_elements_ to account for the removed element.
|
||
|
|
// Release memory ordering prevents the reads and writes to
|
||
|
|
// queue_[next_write_index_] to be reordered to after the decrement. (Once
|
||
|
|
// the decrement has finished, Insert() might start accessing that element.)
|
||
|
|
std::atomic_fetch_sub_explicit(&num_elements_, size_t{1},
|
||
|
|
std::memory_order_release);
|
||
|
|
|
||
|
|
++next_read_index_;
|
||
|
|
if (next_read_index_ == queue_.size()) {
|
||
|
|
next_read_index_ = 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
RTC_DCHECK_LT(next_read_index_, queue_.size());
|
||
|
|
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
|
||
|
|
// Returns the current number of elements in the queue. Since elements may be
|
||
|
|
// concurrently added to the queue, the caller must treat this as a lower
|
||
|
|
// bound, not an exact count.
|
||
|
|
// May only be called by the consumer.
|
||
|
|
size_t SizeAtLeast() const {
|
||
|
|
// Acquire memory ordering ensures that we wait for the producer to finish
|
||
|
|
// inserting any element in progress.
|
||
|
|
return std::atomic_load_explicit(&num_elements_, std::memory_order_acquire);
|
||
|
|
}
|
||
|
|
|
||
|
|
private:
|
||
|
|
// Verify that the queue slots complies with the ItemVerifier test. This
|
||
|
|
// function is not thread-safe and can only be used in the constructors.
|
||
|
|
bool VerifyQueueSlots() {
|
||
|
|
for (const auto& v : queue_) {
|
||
|
|
RTC_DCHECK(queue_item_verifier_(v));
|
||
|
|
}
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
|
||
|
|
// TODO(peah): Change this to use std::function() once we can use C++11 std
|
||
|
|
// lib.
|
||
|
|
QueueItemVerifier queue_item_verifier_;
|
||
|
|
|
||
|
|
// Only accessed by the single producer.
|
||
|
|
size_t next_write_index_ = 0;
|
||
|
|
|
||
|
|
// Only accessed by the single consumer.
|
||
|
|
size_t next_read_index_ = 0;
|
||
|
|
|
||
|
|
// Accessed by both the producer and the consumer and used for synchronization
|
||
|
|
// between them.
|
||
|
|
std::atomic<size_t> num_elements_{0};
|
||
|
|
|
||
|
|
// The elements of the queue are acced by both the producer and the consumer,
|
||
|
|
// mediated by num_elements_. queue_.size() is constant.
|
||
|
|
std::vector<T> queue_;
|
||
|
|
|
||
|
|
SwapQueue(const SwapQueue&) = delete;
|
||
|
|
SwapQueue& operator=(const SwapQueue&) = delete;
|
||
|
|
};
|
||
|
|
|
||
|
|
} // namespace webrtc
|
||
|
|
|
||
|
|
#endif // RTC_BASE_SWAP_QUEUE_H_
|