ringbuffer.cpp

/*
 * Block-free ringbuffer for synchronisation of producer and consumer threads.
 *
 * Audio callback delivers data, the other thread reads data.
 *
 * Size must be a multiple of the real-time buffer size (e.g. JACK buffer).
 * For the non-realtime thread this is not an issue as long as it's large
 * enough to hold at least two consumer frames (theoretical minimum)
 *
 * Caveats: if consumer threads waits too long, producer can overrun the
 * buffer. This may not be a problem, for this we have resync() that puts
 * the consumer readpointer right on top of the consumer writepointer.
 */

#include <iostream>
#include "ringbuffer.h"
#include <unistd.h>
#include <string.h> // memcpy


 /*
  * Size is specified as #items, not bytes. Item type is now float and will
  * eventually be set in template form
  */
RingBuffer::RingBuffer(unsigned long size,std::string name)
{
  tail=0;
  head=0;
  this->size=size;
  itemsize=sizeof(float);
  buffer = new float [size]; // allocate storage
  this->name=name;
  blockingPush=false;
  blockingPop=false;
} // RingBuffer()


RingBuffer::~RingBuffer()
{
  delete [] buffer;
} // ~RingBuffer()


unsigned long RingBuffer::items_available_for_write()
{
long pointerspace=head.load()-tail.load(); // signed

  if(pointerspace > 0) return pointerspace; // NB: > 0 so NOT including 0
  else return pointerspace+size;
} // items_available_for_write()


unsigned long RingBuffer::items_available_for_read()
{
long pointerspace=tail.load()-head.load(); // signed

  if(pointerspace >= 0) return pointerspace; // NB: >= 0 so including 0
  else return pointerspace+size;
} // items_available_for_read()


void RingBuffer::pushMayBlock(bool block)
{
  this->blockingPush=block;
} // pushMayBlock()


void RingBuffer::popMayBlock(bool block)
{
  this->blockingPop=block;
} // popMayBlock()


void RingBuffer::setBlockingNap(unsigned long blockingNap)
{
  this->blockingNap=blockingNap;
} // setBlockingNap()


/*
 * Try to write as many items as possible and return the number actually written
 */
unsigned long RingBuffer::push(float *data,unsigned long n)
{
  unsigned long space=size;

  if(blockingPush){
    while((space=items_available_for_write())<n){ // blocking
      usleep(blockingNap);
    } // while
  } // if
  if(space==0) return 0;
  unsigned long n_to_write = n<=space ? n : space; // limit

  const auto current_tail = tail.load();
  if(current_tail + n_to_write <= size){ // chunk fits without wrapping
    memcpy(buffer+current_tail,data,n_to_write*itemsize);
  }
  else {
    unsigned long first_chunk=size-current_tail;
    memcpy(buffer+current_tail,data,first_chunk*itemsize);
    memcpy(buffer,data+first_chunk,(n_to_write-first_chunk)*itemsize);
  }
  tail.store((current_tail+n_to_write)%size);
  return n_to_write;
} // push()


/*
 * Try to read as many items as possible and return the number actually read
 */
unsigned long RingBuffer::pop(float *data,unsigned long n)
{
  unsigned long space=size;

  if(blockingPop){
    while((space=items_available_for_read())<n){ // blocking
      usleep(blockingNap);
    } // while
  } // if
  if(space==0) return 0;
  unsigned long n_to_read = n<=space ? n : space; // limit

  const auto current_head = head.load();
  if(current_head + n_to_read <= size){ // no wrapping necessary
    memcpy(data,buffer+current_head,n_to_read*itemsize);
  }
  else {
    unsigned long first_chunk=size-current_head;
    memcpy(data,buffer+current_head,first_chunk*itemsize);
    memcpy(data+first_chunk,buffer,(n_to_read-first_chunk)*itemsize);
  }
  head.store((current_head+n_to_read)%size); // zo ongeveer
  return n_to_read;
} // pop()


bool RingBuffer::isLockFree()
{
  return (tail.is_lock_free() && head.is_lock_free());
} // isLockFree()