semi_lockless_fifo.h

/*
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
 
    http://www.apache.org/licenses/LICENSE-2.0
 
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_