ringbuffer.h

// Copyright (c) 2010 The Chromium OS Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//
// Templatized, thread-safe RingBuffer.  Acts like a FIFO, but using a
// fixed-size buffer that wraps around and optimized for operations that
// read or write many elements at a time.  Commonly used to buffer audio
// samples that need to be passed from one thread to another.
//
// Supports a flag "finished" so the writing thread can notify the reading
// thread that there's no more data.
//
// The implementation is all contained within this .h file because
// it's templatized.

#ifndef SPEECH_CLIENT_SYNTHESIS_SERVICE_RINGBUFFER_H_
#define SPEECH_CLIENT_SYNTHESIS_SERVICE_RINGBUFFER_H_

#include "threading.h"

namespace speech_synthesis {

template<typename T> class RingBuffer {
 public:
  // Construct a RingBuffer using a given Threading object (used for
  // locking - may not be NULL) of a given capacity (cannot be increased
  // later).
  RingBuffer(Threading* theading, int capacity);

  // Destructor.
  ~RingBuffer();

  //
  // Methods for either thread
  //

  int GetCapacity() { return capacity_; }

  //
  // Methods for the writer thread
  //

  // Reset to the initial state: the ring buffer is empty and marked as
  // unfinished.
  void Reset();

  // Get the number of items of type T that are available to be written.
  // This will be a number between 0 and capacity, inclusive.
  int WriteAvail();

  // Write *len* elements to the end of the ring buffer.  Returns true on
  // success.  If all *len* elements cannot be written without blocking,
  // returns false and writes nothing.
  bool Write(const T* data, int len);

  // Mark the buffer as finished.  Future write operations will fail.
  // Read operations will succeed until the buffer is empty, but
  // IsFinished() will return true immediately.
  void MarkFinished();

  //
  // Methods for the reader thread
  //

  // Get the number of items of type T that are available to be read.
  // This will be a number between 0 and capacity, inclusive.
  int ReadAvail();

  // Read *len* elements from the front of the ring buffer.  Returns true on
  // success.  If all *len* elements cannot be read without blocking,
  // returns false and reads nothing.
  bool Read(T* data, int len);

  // Returns true if the buffer has been marked as finished by a call to
  // MarkFinished, whether the buffer is empty or not.
  bool IsFinished();

 private:
  Mutex* mutex_;
  T* buffer_;
  bool finished_;
  int capacity_;
  volatile int read_pos_;
  volatile int write_pos_;
};

//
// Implementation notes:
//
// We use a simple mutex to protect all of the fields.
//
// A special value of -1 is used for read_pos_ to indicate that the buffer
// is empty, otherwise there would be no way to distinguish between an empty
// buffer and a full buffer when read_pos_ == write_pos_.
//

template<typename T> RingBuffer<T>::RingBuffer(
    Threading* threading, int capacity) {
  capacity_ = capacity;
  mutex_ = threading->CreateMutex();
  buffer_ = new T[capacity_];
  Reset();
}

template<typename T> RingBuffer<T>::~RingBuffer() {
  delete mutex_;
  delete[] buffer_;
}

template<typename T> void RingBuffer<T>::Reset() {
  mutex_->Lock();
  finished_ = false;
  read_pos_ = -1;
  write_pos_ = 0;
  mutex_->Unlock();
};

template<typename T> int RingBuffer<T>::WriteAvail() {
  int avail = capacity_;
  mutex_->Lock();
  if (read_pos_ != -1) {
    avail = read_pos_ - write_pos_;
  }
  mutex_->Unlock();
  if (avail < 0) {
    avail += capacity_;
  }
  return avail;
}

template<typename T> bool RingBuffer<T>::Write(const T* data, int len) {
  mutex_->Lock();
  if (finished_) {
    mutex_->Unlock();
    return false;
  }
  int avail = capacity_;
  if (read_pos_ != -1) {
    avail = read_pos_ - write_pos_;
  }
  if (avail < 0) {
    avail += capacity_;
  }
  if (len > avail) {
    mutex_->Unlock();
    return false;
  }

  if (read_pos_ == -1) {
    read_pos_ = write_pos_;
  }
  for (int i = 0; i < len; i++) {
    buffer_[write_pos_] = data[i];
    write_pos_ = (write_pos_ + 1) % capacity_;
  }
  mutex_->Unlock();

  return true;
}

template<typename T> void RingBuffer<T>::MarkFinished() {
  mutex_->Lock();
  finished_ = true;
  mutex_->Unlock();
}

template<typename T> int RingBuffer<T>::ReadAvail() {
  mutex_->Lock();
  if (read_pos_ == -1) {
    mutex_->Unlock();
    return 0;
  }
  int avail = write_pos_ - read_pos_;
  mutex_->Unlock();
  if (avail <= 0) {
    avail += capacity_;
  }
  return avail;
}

template<typename T> bool RingBuffer<T>::Read(T* data, int len) {
  mutex_->Lock();
  int avail;
  if (read_pos_ == -1) {
    avail = 0;
  } else {
    avail = write_pos_ - read_pos_;
    if (avail <= 0) {
      avail += capacity_;
    }
  }
  if (len > avail) {
    mutex_->Unlock();
    return false;
  }

  for (int i = 0; i < len; i++) {
    data[i] = buffer_[read_pos_];
    read_pos_ = (read_pos_ + 1) % capacity_;
  }

  if (read_pos_ == write_pos_) {
    read_pos_ = -1;
  }

  mutex_->Unlock();

  return true;
}

template<typename T> bool RingBuffer<T>::IsFinished() {
  mutex_->Lock();
  bool is_finished = finished_;
  mutex_->Unlock();
  return is_finished;
}

}  // namespace speech_synthesis

#endif  // SPEECH_CLIENT_SYNTHESIS_SERVICE