/* The write pointer, represented as an offset into the array. */ privateint offset = 0;
/* The read pointer is encoded implicitly by keeping track of the number of * unconsumed elements. We can then determine its position by backing up * that many positions before the read position. */ privateint unconsumedElements = 0;
/** * Constructs a new RingBuffer with the specified capacity, which must be * positive. * * @param size The capacity of the new ring buffer. * @throws IllegalArgumentException If the capacity is negative. */ @SuppressWarnings("unchecked") public RingBuffer(int size) { /* Validate the size. */ if (size <= 0) thrownew IllegalArgumentException("RingBuffer capacity must be positive.");
/* Construct the array to be that size. */ elements = (T[]) new Object[size]; }
/** * Appends an element to the ring buffer, blocking until space becomes * available. * * @param elem The element to add to the ring buffer. * @throws InterruptedException If the thread is interrupted before the * insertion completes. */ publicsynchronizedvoid add(T elem) throws InterruptedException { /* Block until the capacity is nonzero. Otherwise we don't have any * space to write. */ while (unconsumedElements == elements.length) wait();
/* Write the element into the next open spot, then advance the write * pointer forward a step. */ elements[offset] = elem; offset = (offset + 1) % elements.length;
/* Increase the number of unconsumed elements by one, then notify any * threads that are waiting that more data is now available. */ ++unconsumedElements; notifyAll(); }
/** * Returns the maximum capacity of the ring buffer. * * @return The maximum capacity of the ring buffer. */ publicint capacity() { return elements.length; }
/** * Observes, but does not dequeue, the next available element, blocking * until data becomes available. * * @return The next available element. * @throws InterruptedException If the caller is interrupted before data * becomes available. */ publicsynchronized T peek() throws InterruptedException { /* Wait for data to become available. */ while (unconsumedElements == 0) wait();
/* Hand back the next value. The index of this next value is a bit * tricky to compute. We know that there are unconsumedElements * elements waiting to be read, and they're contiguously before the * write position. However, the buffer wraps around itself, and so we * can't just do a naive subtraction; that might end up giving us a * negative index. To avoid this, we'll use a clever trick in which * we'll add to the index the capacity minus the distance. This value * must be positive, since the distance is never greater than the * capacity, and if we then wrap this value around using the modulus * operator we'll end up with a valid index. All of this machinery * works because * * (x + (n - k)) mod n == (x - k) mod n * * And Java's modulus operator works best on positive values. */ return elements[(offset + (capacity() - unconsumedElements)) % capacity()]; }
/** * Removes and returns the next available element, blocking until data * becomes available. * * @return The next available element * @throws InterruptedException If the caller is interrupted before data * becomes available. */ publicsynchronized T remove() throws InterruptedException { /* Use peek() to get the element to return. */ T result = peek();
/* Mark that one fewer elements are now available to read. */ --unconsumedElements;
/* Because there is more space left, wake up any waiting threads. */ notifyAll();
return result; }
/** * Returns the number of elements that are currently being stored in the * ring buffer. * * @return The number of elements currently stored in the ring buffer. */ publicsynchronizedint size() { return unconsumedElements; }
/** Array index of first (oldest) buffer element */
private transient int start = 0;
* Index mod maxElements of the array position following the last buffer
* element. Buffer elements start at elements[start] and "wrap around"
* elements[maxElements-1], ending at elements[decrement(end)].
* For example, elements = {c,a,b}, start=1, end=1 corresponds to
* the buffer [a,b,c].
*/
private transient int end = 0;
/** Flag to indicate if the buffer is currently full. */
private transient boolean full = false;
/** Capacity of the buffer */
private final int maxElements;
public int size() {
int size = 0;
return size;
}
public boolean isEmpty() {
return size() == 0;
}
public boolean isFull() {
return size() == maxElements;
}
public boolean add(Object element) {
if (null == element) {
throw new NullPointerException("Attempted to add null object to buffer");
}
if (full) {
throw new BufferOverflowException("The buffer cannot hold more than " + maxElements + " objects.");
}
elements[end++] = element;
if (end >= maxElements) {
end = 0;
}
if (end == start) {
full = true;
}
return true;
}
public Object get() {
if (isEmpty()) {
throw new BufferUnderflowException("The buffer is already empty");
}
return elements[start];
}
public Object remove() {
if (isEmpty()) {
throw new BufferUnderflowException("The buffer is already empty");
}
Object element = elements[start];
if (null != element) { //\\
elements[start++] = null;
if (start >= maxElements) {
start = 0;
}
full = false;
}
return element;
}
private int increment(int index) {
index++;
if (index >= maxElements) {
index = 0;
}
return index;
}
private int decrement(int index) {
index--;
if (index < 0) {
index = maxElements - 1;
}
return index;
}
public Iterator iterator() {
return new Iterator() {
private int index = start;
private int lastReturnedIndex = -1;
private boolean isFirst = full;
public boolean hasNext() {
return isFirst || (index != end);
}
public Object next() {
if (!hasNext()) {
throw new NoSuchElementException();
}
isFirst = false;
lastReturnedIndex = index;
index = increment(index);
return elements[lastReturnedIndex];
}
public void remove() {
if (lastReturnedIndex == -1) {
throw new IllegalStateException();
}
// First element can be removed quickly
if (lastReturnedIndex == start) {
BoundedFifoBuffer.this.remove();
lastReturnedIndex = -1;
return;
}
int pos = lastReturnedIndex + 1;
if (start < lastReturnedIndex && pos < end) {
// shift in one part
System.arraycopy(elements, pos, elements,
lastReturnedIndex, end - pos);
} else {
// Other elements require us to shift the subsequent elements
while (pos != end) {
if (pos >= maxElements) {
elements[pos - 1] = elements[0];
pos = 0;
} else {
elements[decrement(pos)] = elements[pos];
pos = increment(pos);
}
}
}
lastReturnedIndex = -1;
end = decrement(end);
elements[end] = null;
full = false;
index = decrement(index);
}
Ring buffers are especially useful when you need a hard upper bound for how much data can be in the ring buffer (in the queue).
If you don't want an upper bound on your queue structure, perhaps you should be using a linked list, or a ring buffer that can resize itself (allocate a new, bigger array when it is full, and move all elements to that array).
One way to keep track of write position, read position and the number of elements in the ring buffer is to use a write position and a fill count. The write position marks the next position in the ring buffer to write an element to. The fill count tells how many elements are currently stored in the buffer.
When writing elements to the ring buffer it just has to check the fill count to see if it is full or not. If it is not full the element can be inserted at the write position and the write position advanced to the next free slot.
Similarly, when reading elements from the ring buffer it just has to check the fill count to see if the ring buffer is empty. The read position is calculated by subtracting the fill count from the write position. If the result of that calculation is negative, the write position has wrapped around, and you need to add the size of the buffer (array) to the read position to get the correct read position.
Below is a ring buffer implementation that uses a fill count. Notice that it does not track the read position explicitly, but calculates the read position based on the write position and the fill count. Notice also, that the fill count field is called available (not fillCount).
public class RingBufferFillCount {
public Object[] elements = null;
private int capacity = 0;
private int writePos = 0;
private int available = 0;
public RingBufferFillCount(int capacity) {
this.capacity = capacity;
this.elements = new Object[capacity];
}
public void reset() {
this.writePos = 0;
this.available = 0;
}
public int capacity() { return this.capacity; }
public int available(){ return this.available; }
public int remainingCapacity() {
return this.capacity - this.available;
}
public boolean put(Object element){
if(available < capacity){
if(writePos >= capacity){
writePos = 0;
}
elements[writePos] = element;
writePos++;
available++;
return true;
}
return false;
}
public Object take() {
if(available == 0){
return null;
}
int nextSlot = writePos - available;
if(nextSlot < 0){
nextSlot += capacity;
}
Object nextObj = elements[nextSlot]; elements[nextSlot]=null;
available--;
return nextObj;
}
}
Ring Buffer Using Flip Marker
Another option to keep track of read position, write position and how many elements are in the buffer, is to simply keep a read position in addition to the write position.
How many elements are in the buffer is calculated based on the write and read position. How the calculation looks depends on whether the write position has flipped (wrapped around) or not.
If the write position has not wrapped around you can simply subtract the read position from the write position to know how many elements are in the buffer. If the write position has wrapped around (flipped) then the available space is equal to the capacity minus the read position plus the write position.
To keep track of whether the write position has flipped or not a special "flip marker" is used. That is where the name of the implementation comes from. Actually, in most cases you could just check if the write position is larger or smaller than the read position to detect if the write position has wrapped around. But, that doesn't work when write position and read position are equal (the ring buffer is either completely full or completely empty).
public class RingBufferFlipMarker {
public Object[] elements = null;
public int capacity = 0;
public int writePos = 0;
public int readPos = 0;
public boolean flipped = false; //the flip marker
public RingBufferFlipMarker(int capacity) {
this.capacity = capacity;
this.elements = new Object[capacity];
}
public void reset() {
this.writePos = 0;
this.readPos = 0;
this.flipped = false;
}
public int available() {
if(!flipped){
return writePos - readPos;
}
return capacity - readPos + writePos;
}
public int remainingCapacity() {
if(!flipped){
return capacity - writePos;
}
return readPos - writePos;
}
public boolean put(Object element){
if(!flipped){
if(writePos == capacity){
writePos = 0;
flipped = true;
if(writePos < readPos){
elements[writePos++] = element;
return true;
} else {
return false;
}
} else {
elements[writePos++] = element;
return true;
}
} else {
if(writePos < readPos ){
elements[writePos++] = element;
return true;
} else {
return false;
}
}
}
public Object take() {
if(!flipped){
if(readPos < writePos){
return elements[readPos++];
} else {
return null;
}
} else {
if(readPos == capacity){
readPos = 0;
flipped = false;
if(readPos < writePos){
return elements[readPos++];
} else {
return null;
}
} else {
return elements[readPos++];
}
}
}
}
Ring Buffer Using Fill Count - Including Batch Operations
public class RingBufferFillCount {
public Object[] elements = null;
public int capacity = 0;
public int writePos = 0;
public int available = 0;
public RingBufferFillCount(int capacity) {
this.capacity = capacity;
this.elements = new Object[capacity];
}
public void reset() {
this.writePos = 0;
this.available = 0;
}
public int remainingCapacity() {
return this.capacity - this.available;
}
public boolean put(Object element){
if(available < capacity){
if(writePos >= capacity){
writePos = 0;
}
elements[writePos] = element;
writePos++;
available++;
return true;
}
return false;
}
public int put(Object[] newElements){
return put(newElements, newElements.length);
}
public int put(Object[] newElements, int length){
int readPos = 0;
if(this.writePos > this.available){
//space above writePos is all empty
if(length <= this.capacity - this.writePos){
//space above writePos is sufficient to insert batch
for(; readPos < length; readPos++){
this.elements[this.writePos++] = newElements[readPos];
}
this.available += readPos;
return length;
} else {
//both space above writePos and below writePos is necessary to use
//to insert batch.
int lastEmptyPos = writePos - available;
for(; this.writePos < this.capacity; this.writePos++){
this.elements[this.writePos] = newElements[readPos++];
}
//fill into bottom of array too.
this.writePos = 0;
int endPos = Math.min(length - readPos, capacity - available - readPos);
for(;this.writePos < endPos; this.writePos++){
this.elements[this.writePos] = newElements[readPos++];
}
this.available += readPos;
return readPos;
}
} else {
int endPos = this.capacity - this.available + this.writePos;
for(; this.writePos < endPos; this.writePos++){
this.elements[this.writePos] = newElements[readPos++];
}
this.available += readPos;
return readPos;
}
}
public Object take() {
if(available == 0){
return null;
}
int nextSlot = writePos - available;
if(nextSlot < 0){
nextSlot += capacity;
}
Object nextObj = elements[nextSlot];
available--;
return nextObj;
}
public int take(Object[] into){
return take(into, into.length);
}
public int take(Object[] into, int length){
int intoPos = 0;
if(available <= writePos){
int nextPos= writePos - available;
int endPos = nextPos + Math.min(available, length);
for(;nextPos < endPos; nextPos++){
into[intoPos++] = this.elements[nextPos];
}
this.available -= intoPos;
return intoPos;
} else {
int nextPos = writePos - available + capacity;
int leftInTop = capacity - nextPos;
if(length <= leftInTop){
//copy directly
for(; intoPos < length; intoPos++){
into[intoPos] = this.elements[nextPos++];
}
this.available -= length;
return length;
} else {
//copy top
for(; nextPos < capacity; nextPos++){
into[intoPos++] = this.elements[nextPos];
}
//copy bottom - from 0 to writePos
nextPos = 0;
int leftToCopy = length - intoPos;
int endPos = Math.min(writePos, leftToCopy);
for(;nextPos < endPos; nextPos++){
into[intoPos++] = this.elements[nextPos];
}
this.available -= intoPos;
return intoPos;
}
}
}
}
Ring Buffer Using Flip Marker - Including Batch Operations
public class RingBufferFlip {
public Object[] elements = null;
public int capacity = 0;
public int writePos = 0;
public int readPos = 0;
public boolean flipped = false;
public RingBufferFlip(int capacity) {
this.capacity = capacity;
this.elements = new Object[capacity];
}
public void reset() {
this.writePos = 0;
this.readPos = 0;
this.flipped = false;
}
public int available() {
if(!flipped){
return writePos - readPos;
}
return capacity - readPos + writePos;
}
public int remainingCapacity() {
if(!flipped){
return capacity - writePos;
}
return readPos - writePos;
}
public boolean put(Object element){
if(!flipped){
if(writePos == capacity){
writePos = 0;
flipped = true;
if(writePos < readPos){
elements[writePos++] = element;
return true;
} else {
return false;
}
} else {
elements[writePos++] = element;
return true;
}
} else {
if(writePos < readPos ){
elements[writePos++] = element;
return true;
} else {
return false;
}
}
}
public int put(Object[] newElements, int length){
int newElementsReadPos = 0;
if(!flipped){
//readPos lower than writePos - free sections are:
//1) from writePos to capacity
//2) from 0 to readPos
if(length <= capacity - writePos){
//new elements fit into top of elements array - copy directly
for(; newElementsReadPos < length; newElementsReadPos++){
this.elements[this.writePos++] = newElements[newElementsReadPos];
}
return newElementsReadPos;
} else {
//new elements must be divided between top and bottom of elements array
//writing to top
for(;this.writePos < capacity; this.writePos++){
this.elements[this.writePos] = newElements[newElementsReadPos++];
}
//writing to bottom
this.writePos = 0;
this.flipped = true;
int endPos = Math.min(this.readPos, length - newElementsReadPos);
for(; this.writePos < endPos; this.writePos++){
this.elements[writePos] = newElements[newElementsReadPos++];
}
return newElementsReadPos;
}
} else {
//readPos higher than writePos - free sections are:
//1) from writePos to readPos
int endPos = Math.min(this.readPos, this.writePos + length);
for(; this.writePos < endPos; this.writePos++){
this.elements[this.writePos] = newElements[newElementsReadPos++];
}
return newElementsReadPos;
}
}
public Object take() {
if(!flipped){
if(readPos < writePos){
return elements[readPos++];
} else {
return null;
}
} else {
if(readPos == capacity){
readPos = 0;
flipped = false;
if(readPos < writePos){
return elements[readPos++];
} else {
return null;
}
} else {
return elements[readPos++];
}
}
}
public int take(Object[] into, int length){
int intoWritePos = 0;
if(!flipped){
//writePos higher than readPos - available section is writePos - readPos
int endPos = Math.min(this.writePos, this.readPos + length);
for(; this.readPos < endPos; this.readPos++){
into[intoWritePos++] = this.elements[this.readPos];
}
return intoWritePos;
} else {
//readPos higher than writePos - available sections are
//top + bottom of elements array
if(length <= capacity - readPos){
//length is lower than the elements available at the top
//of the elements array - copy directly
for(; intoWritePos < length; intoWritePos++){
into[intoWritePos] = this.elements[this.readPos++];
}
return intoWritePos;
} else {
//length is higher than elements available at the top of the elements array
//split copy into a copy from both top and bottom of elements array.
//copy from top
for(; this.readPos < capacity; this.readPos++){
into[intoWritePos++] = this.elements[this.readPos];
}
//copy from bottom
this.readPos = 0;
this.flipped = false;
int endPos = Math.min(this.writePos, length - intoWritePos);
for(; this.readPos < endPos; this.readPos++){
into[intoWritePos++] = this.elements[this.readPos];
}
return intoWritePos;
}
}
}
}
public class ArrayQueue implements Queue
{
public double[] buf; // Circular buffer
public int read, write; // read and write pointers // Constructor
public ArrayQueue(int size)
{
buf = new double[size]; // Create array for circular buffer
read = 0; // Initialized read & write pointers
write = 0;
}
/* ====================================================
enqueue(x ):
==================================================== */
public void enqueue( double x ) throws Exception
{
if ( read == ( write + 1 ) % (buf.length) ) // Full...
{
throw new Exception("Queue is full");
}
buf[write] = x; // Store x in buf at write pointer
write = (write+1)%(buf.length); // Advance the write pointer
}
/* ====================================================
dequeue():
==================================================== */
public double dequeue( ) throws Exception
{
double r; // Variable used to save the return value
if ( read == write )
{
throw new Exception("Queue is empty");
}
r = buf[read]; // Save return value
read = (read+1)%(buf.length); // Advance the read pointer
return r;
}
}
publicclassNumberFixedLengthFifoQueueimplementsQueue<Number>{protectedNumber[]ring;protectedintindex;/**
* @param initialValues contains the ring's initial values.
* The "oldest" value in the queue is expected to reside in
* position 0, the newest one in position length-1.
*/publicNumberFixedLengthFifoQueue(Number[]initialValues){// This is a little ugly, but there are no// generic arrays in Javaring=newNumber[initialValues.length];// We don't want to work on the original dataSystem.arraycopy(initialValues,0,ring,0,initialValues.length);// The next time we add something to the queue,// the oldest element should be replacedindex=0;}@Overridepublicbooleanadd(Numbernewest){returnoffer(newest);}@OverridepublicNumberelement(){returnring[getHeadIndex()];}@Overridepublicbooleanoffer(Numbernewest){Numberoldest=ring[index];ring[index]=newest;incrIndex();returntrue;}@OverridepublicNumberpeek(){returnring[getHeadIndex()];}@OverridepublicNumberpoll(){thrownewIllegalStateException("The poll method is not available for NumberFixedLengthFifoQueue.");}@OverridepublicNumberremove(){thrownewIllegalStateException("The remove method is not available for NumberFixedLengthFifoQueue.");}@OverridepublicNumberget(intabsIndex)throwsIndexOutOfBoundsException{if(absIndex>=ring.length){thrownewIndexOutOfBoundsException("Invalid index "+absIndex);}inti=index+absIndex;if(i>=ring.length){i-=ring.length;}returnring[i];}@OverridepublicStringtoString(){StringBuffersb=newStringBuffer("[");for(inti=index,n=0;n<ring.length;i=nextIndex(i),n++){sb.append(ring[i]);if(n+1<ring.length){sb.append(", ");}}returnsb.append("]").toString();}protectedvoidincrIndex(){index=nextIndex(index);}protectedintnextIndex(intcurrent){if(current+1>=ring.length){return0;}elsereturncurrent+1;}protectedintpreviousIndex(intcurrent){if(current-1<0){returnring.length-1;}elsereturncurrent-1;}protectedintgetHeadIndex(){if(index==0){returnring.length-1;}elsereturnindex-1;}}
The queue "rolls" through the ring. Adding a new element at the head of the queue automatically removes the oldest element in the queue - no copying of arrays or resetting of object references required. Unlike with linked lists we can actually access each element in the ring directly with the get method. Finally, we can create a subclass of our queue object which will graciously roll over as new values are added into the queue/ring.
public class ArrayDeque<E> extends AbstractCollection<E>
implements Deque<E>, Cloneable, Serializable
{
transient Object[] elements; // non-private to simplify nested class access
// The index of the element at the head of the deque (which is the
// element that would be removed by remove() or pop()); or an
// arbitrary number equal to tail if the deque is empty.)
transient int head;
// The index at which the next element would be added to the tail
transient int tail;
public void addFirst(E e) {
if (e == null)
throw new NullPointerException();
elements[head = (head - 1) & (elements.length - 1)] = e;
if (head == tail)
doubleCapacity();
}
public void addLast(E e) {
if (e == null)
throw new NullPointerException();
elements[tail] = e;
if ( (tail = (tail + 1) & (elements.length - 1)) == head)
doubleCapacity();
}
public E pollFirst() {
int h = head;
@SuppressWarnings("unchecked")
E result = (E) elements[h];
// Element is null if deque empty
if (result == null)
return null;
elements[h] = null; // Must null out slot
head = (h + 1) & (elements.length - 1);
return result;
}
public E pollLast() {
int t = (tail - 1) & (elements.length - 1);
@SuppressWarnings("unchecked")
E result = (E) elements[t];
if (result == null)
return null;
elements[t] = null;
tail = t;
return result;
}
public E getFirst() {
@SuppressWarnings("unchecked")
E result = (E) elements[head];
if (result == null)
throw new NoSuchElementException();
return result;
}
public E getLast() {
@SuppressWarnings("unchecked")
E result = (E) elements[(tail - 1) & (elements.length - 1)];
if (result == null)
throw new NoSuchElementException();
return result;
}
}
