producer_consumer_queue.h

// Copyright 2014 Google Inc. All rights reserved.
//
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file or at
// https://developers.google.com/open-source/licenses/bsd
 
#ifndef PACKAGER_MEDIA_BASE_PRODUCER_CONSUMER_QUEUE_H_
#define PACKAGER_MEDIA_BASE_PRODUCER_CONSUMER_QUEUE_H_
 
#include <deque>
 
#include "packager/base/strings/stringprintf.h"
#include "packager/base/synchronization/condition_variable.h"
#include "packager/base/synchronization/lock.h"
#include "packager/base/timer/elapsed_timer.h"
#include "packager/status.h"
 
namespace shaka {
namespace media {
 
static const size_t kUnlimitedCapacity = 0u;
static const int64_t kInfiniteTimeout = -1;
 
template <class T>
class ProducerConsumerQueue {
 public:
  explicit ProducerConsumerQueue(size_t capacity);
 
  ProducerConsumerQueue(size_t capacity, size_t starting_pos);
 
  ~ProducerConsumerQueue();
 
  Status Push(const T& element, int64_t timeout_ms);
 
  Status Pop(T* element, int64_t timeout_ms);
 
  Status Peek(size_t pos, T* element, int64_t timeout_ms);
 
  void Stop() {
    base::AutoLock l(lock_);
    stop_requested_ = true;
    not_empty_cv_.Broadcast();
    not_full_cv_.Broadcast();
    new_element_cv_.Broadcast();
  }
 
  bool Empty() const {
    base::AutoLock l(lock_);
    return q_.empty();
  }
 
  size_t Size() const {
    base::AutoLock l(lock_);
    return q_.size();
  }
 
  size_t HeadPos() const {
    base::AutoLock l(lock_);
    return head_pos_;
  }
 
  size_t TailPos() const {
    base::AutoLock l(lock_);
    return head_pos_ + q_.size() - 1;
  }
 
  bool Stopped() const {
    base::AutoLock l(lock_);
    return stop_requested_;
  }
 
 private:
  // Move head_pos_ to center on pos.
  void SlideHeadOnCenter(size_t pos);
 
  const size_t capacity_;  // Maximum number of elements; zero means unlimited.
  mutable base::Lock lock_;  // Lock protecting all other variables below.
  size_t head_pos_;          // Head position.
  std::deque<T> q_;          // Internal queue holding the elements.
  base::ConditionVariable not_empty_cv_;
  base::ConditionVariable not_full_cv_;
  base::ConditionVariable new_element_cv_;
  bool stop_requested_;  // True after Stop has been called.
 
  DISALLOW_COPY_AND_ASSIGN(ProducerConsumerQueue);
};
 
// Implementations of non-inline functions.
template <class T>
ProducerConsumerQueue<T>::ProducerConsumerQueue(size_t capacity)
    : capacity_(capacity),
      head_pos_(0),
      not_empty_cv_(&lock_),
      not_full_cv_(&lock_),
      new_element_cv_(&lock_),
      stop_requested_(false) {}
 
template <class T>
ProducerConsumerQueue<T>::ProducerConsumerQueue(size_t capacity,
                                                size_t starting_pos)
    : capacity_(capacity),
      head_pos_(starting_pos),
      not_empty_cv_(&lock_),
      not_full_cv_(&lock_),
      new_element_cv_(&lock_),
      stop_requested_(false) {
}
 
template <class T>
ProducerConsumerQueue<T>::~ProducerConsumerQueue() {}
 
template <class T>
Status ProducerConsumerQueue<T>::Push(const T& element, int64_t timeout_ms) {
  base::AutoLock l(lock_);
  bool woken = false;
 
  // Check for queue shutdown.
  if (stop_requested_)
    return Status(error::STOPPED, "");
 
  base::ElapsedTimer timer;
  base::TimeDelta timeout_delta = base::TimeDelta::FromMilliseconds(timeout_ms);
 
  if (capacity_) {
    while (q_.size() == capacity_) {
      if (timeout_ms < 0) {
        // Wait forever, or until Stop.
        not_full_cv_.Wait();
      } else {
        base::TimeDelta elapsed = timer.Elapsed();
        if (elapsed < timeout_delta) {
          // Wait with timeout, or until Stop.
          not_full_cv_.TimedWait(timeout_delta - elapsed);
        } else {
          // We're through waiting.
          return Status(error::TIME_OUT, "Time out on pushing.");
        }
      }
      // Re-check for queue shutdown after waking from Wait.
      if (stop_requested_)
        return Status(error::STOPPED, "");
 
      woken = true;
    }
    DCHECK_LT(q_.size(), capacity_);
  }
 
  // Signal consumer to proceed if we are going to create some elements.
  if (q_.empty())
    not_empty_cv_.Signal();
  new_element_cv_.Signal();
 
  q_.push_back(element);
 
  // Signal other producers if we just acquired more capacity.
  if (woken && q_.size() != capacity_)
    not_full_cv_.Signal();
  return Status::OK;
}
 
template <class T>
Status ProducerConsumerQueue<T>::Pop(T* element, int64_t timeout_ms) {
  base::AutoLock l(lock_);
  bool woken = false;
 
  base::ElapsedTimer timer;
  base::TimeDelta timeout_delta = base::TimeDelta::FromMilliseconds(timeout_ms);
 
  while (q_.empty()) {
    if (stop_requested_)
      return Status(error::STOPPED, "");
 
    if (timeout_ms < 0) {
      // Wait forever, or until Stop.
      not_empty_cv_.Wait();
    } else {
      base::TimeDelta elapsed = timer.Elapsed();
      if (elapsed < timeout_delta) {
        // Wait with timeout, or until Stop.
        not_empty_cv_.TimedWait(timeout_delta - elapsed);
      } else {
        // We're through waiting.
        return Status(error::TIME_OUT, "Time out on popping.");
      }
    }
    woken = true;
  }
 
  // Signal producer to proceed if we are going to create some capacity.
  if (q_.size() == capacity_)
    not_full_cv_.Signal();
 
  *element = q_.front();
  q_.pop_front();
  ++head_pos_;
 
  // Signal other consumers if we have more elements.
  if (woken && !q_.empty())
    not_empty_cv_.Signal();
  return Status::OK;
}
 
template <class T>
Status ProducerConsumerQueue<T>::Peek(size_t pos,
                                      T* element,
                                      int64_t timeout_ms) {
  base::AutoLock l(lock_);
  if (pos < head_pos_) {
    return Status(
        error::INVALID_ARGUMENT,
        base::StringPrintf(
            "pos (%zu) is too small; head is at %zu.", pos, head_pos_));
  }
 
  bool woken = false;
 
  base::ElapsedTimer timer;
  base::TimeDelta timeout_delta = base::TimeDelta::FromMilliseconds(timeout_ms);
 
  // Move head to create some space (move the sliding window centered @ pos).
  SlideHeadOnCenter(pos);
 
  while (pos >= head_pos_ + q_.size()) {
    if (stop_requested_)
      return Status(error::STOPPED, "");
 
    if (timeout_ms < 0) {
      // Wait forever, or until Stop.
      new_element_cv_.Wait();
    } else {
      base::TimeDelta elapsed = timer.Elapsed();
      if (elapsed < timeout_delta) {
        // Wait with timeout, or until Stop.
        new_element_cv_.TimedWait(timeout_delta - elapsed);
      } else {
        // We're through waiting.
        return Status(error::TIME_OUT, "Time out on peeking.");
      }
    }
    // Move head to create some space (move the sliding window centered @ pos).
    SlideHeadOnCenter(pos);
    woken = true;
  }
 
  *element = q_[pos - head_pos_];
 
  // Signal other consumers if we have more elements.
  if (woken && !q_.empty())
    new_element_cv_.Signal();
  return Status::OK;
}
 
template <class T>
void ProducerConsumerQueue<T>::SlideHeadOnCenter(size_t pos) {
  lock_.AssertAcquired();
 
  if (capacity_) {
    // Signal producer to proceed if we are going to create some capacity.
    if (q_.size() == capacity_ && pos > head_pos_ + capacity_ / 2)
      not_full_cv_.Signal();
 
    while (!q_.empty() && pos > head_pos_ + capacity_ / 2) {
      ++head_pos_;
      q_.pop_front();
    }
  }
}
 
}  // namespace media
}  // namespace shaka
 
#endif  // PACKAGER_MEDIA_BASE_PRODUCER_CONSUMER_QUEUE_H_