Revision control
Copy as Markdown
/*
Copyright 2018 Google Inc. All Rights Reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS-IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
#ifndef RESONANCE_AUDIO_UTILS_SEMI_LOCKLESS_FIFO_H_
#define RESONANCE_AUDIO_UTILS_SEMI_LOCKLESS_FIFO_H_
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <vector>
#include "base/logging.h"
namespace vraudio {
// Thread-safe multiple producer - single consumer FIFO queue to share data
// between threads. The FIFO takes over ownership of the queue elements. Note
// that |PushBack| calls are synchronized with a mutex and may block. Calls to
// |PopFront| are lockless and never block.
//
// @tparam DataType Object type that the FIFO handles.
template <typename DataType>
class SemiLocklessFifo {
public:
typedef std::chrono::steady_clock::duration ClockDuration;
SemiLocklessFifo();
~SemiLocklessFifo();
// Takes over ownership of |input| and pushes it to the FIFO queue back.
//
// @param input Input element to be added to the FIFO queue.
void PushBack(DataType&& input);
// Pops element from FIFO queue front.
//
// @return Element from FIFO queue front. Must not be called if the queue is
// empty.
DataType PopFront();
// Returns true if FIFO queue is empty, false otherwise. This method is *not*
// thread-safe and should only be called from the consumer thread.
bool Empty() const;
// Clears the FIFO queue and deletes all its elements. This method is *not*
// thread-safe and should only be called from the consumer thread.
void Clear();
// Sleeps until the number of elements in the FIFO queue drop below a target
// threshold. This method can be used to synchronize the producer and the
// consumer. Sleeping is enabled by default and can be disabled via
// |EnableBlockingSleepUntilMethods|.
//
// @param target_size Target size of FIFO queue.
// @param max_wait Maximum waiting period.
// @return True if number of FIFO elements is below target size.
bool SleepUntilBelowSizeTarget(size_t target_size,
const ClockDuration& max_wait);
// Sleeps until the number of elements in the FIFO queue is greater or equal a
// target threshold. This method can be used to synchronize the producer and
// the consumer. Sleeping is enabled by default and can be disabled via
// |EnableBlockingSleepUntilMethods|.
//
// @param target_size Target size of FIFO queue.
// @param max_wait Maximum waiting period.
// @return True if number of FIFO elements is greater or equal the target
// size.
bool SleepUntilNumElementsInQueue(size_t target_size,
const ClockDuration& max_wait);
// Allows for unblocking |SleepUntil[BelowSizeTarget|NumElementsInQueue]|
// method.
void EnableBlockingSleepUntilMethods(bool enable);
private:
// Node in single-linked list.
struct Node {
Node() : next(nullptr) {}
std::atomic<Node*> next;
DataType data;
};
// Head of linked list.
Node* head_;
// Tail of linked list.
Node* tail_;
// Number of elements.
std::atomic<size_t> fifo_size_;
// Mutex to synchronize |PushBack| calls from multiple threads.
std::mutex push_mutex_;
// Conditional to signal consumption.
std::condition_variable pop_conditional_;
// Mutex to block on until signal consumption occurs.
std::mutex pop_conditional_mutex_;
// Conditional to signal new elements on the FIFO.
std::condition_variable push_conditional_;
// Mutex to block on until new elements have been added to the FIFO.
std::mutex push_conditional_mutex_;
// Flag to enable and disable blocking sleeping calls.
std::atomic<bool> enable_sleeping_;
};
template <typename DataType>
SemiLocklessFifo<DataType>::SemiLocklessFifo()
: fifo_size_(0), enable_sleeping_(true) {
head_ = tail_ = new Node();
}
template <typename DataType>
SemiLocklessFifo<DataType>::~SemiLocklessFifo() {
Clear();
DCHECK_EQ(head_, tail_);
DCHECK(head_->next.load() == nullptr);
delete head_;
}
template <typename DataType>
void SemiLocklessFifo<DataType>::PushBack(DataType&& input) {
std::lock_guard<std::mutex> lock(push_mutex_);
tail_->data = std::move(input);
Node* const new_node = new Node();
DCHECK(tail_->next.load() == nullptr);
tail_->next = new_node;
tail_ = new_node;
++fifo_size_;
{
// Taking the lock and dropping it immediately assure that the notify
// cannot happen between the check of the predicate and wait of the
// |push_conditional_|.
std::lock_guard<std::mutex> lock(push_conditional_mutex_);
}
push_conditional_.notify_all();
}
template <typename DataType>
DataType SemiLocklessFifo<DataType>::PopFront() {
DCHECK(!Empty());
Node* const front_node = head_;
head_ = front_node->next;
DataType output = std::move(front_node->data);
delete front_node;
DCHECK_GT(fifo_size_.load(), 0u);
--fifo_size_;
{
// Taking the lock and dropping it immediately assure that the notify
// cannot happen between the check of the predicate and wait of the
// |pop_conditional_|.
std::lock_guard<std::mutex> lock(pop_conditional_mutex_);
}
pop_conditional_.notify_one();
return output;
}
template <typename DataType>
bool SemiLocklessFifo<DataType>::Empty() const {
return fifo_size_.load() == 0;
}
template <typename DataType>
void SemiLocklessFifo<DataType>::Clear() {
while (!Empty()) {
PopFront();
}
DCHECK_EQ(fifo_size_, 0u);
}
template <typename DataType>
bool SemiLocklessFifo<DataType>::SleepUntilBelowSizeTarget(
size_t target_size, const ClockDuration& max_wait) {
DCHECK_GT(target_size, 0);
std::unique_lock<std::mutex> lock(pop_conditional_mutex_);
pop_conditional_.wait_for(lock, max_wait, [this, target_size]() {
return fifo_size_ < target_size || !enable_sleeping_.load();
});
return fifo_size_ < target_size;
}
template <typename DataType>
bool SemiLocklessFifo<DataType>::SleepUntilNumElementsInQueue(
size_t target_size, const ClockDuration& max_wait) {
DCHECK_GT(target_size, 0u);
std::unique_lock<std::mutex> lock(push_conditional_mutex_);
push_conditional_.wait_for(lock, max_wait, [this, target_size]() {
return fifo_size_ >= target_size || !enable_sleeping_.load();
});
return fifo_size_ >= target_size;
}
template <typename DataType>
void SemiLocklessFifo<DataType>::EnableBlockingSleepUntilMethods(bool enable) {
enable_sleeping_ = enable;
// Taking the lock and dropping it immediately assure that the notify
// cannot happen between the check of the predicate and wait of the
// |pop_conditional_| and |push_conditional_|.
{ std::lock_guard<std::mutex> lock(pop_conditional_mutex_); }
{ std::lock_guard<std::mutex> lock(push_conditional_mutex_); }
pop_conditional_.notify_one();
push_conditional_.notify_one();
}
} // namespace vraudio
#endif // RESONANCE_AUDIO_UTILS_SEMI_LOCKLESS_FIFO_H_