This is a data structure I wrote:
- It is an circular first-in-first-out queue (a ringbuffer);
- It has batched removal/popping - it returns an array of a fixed buffer size;
- It is unbounded, and grows itself as needed;
- It is generics-compliant (at the cost of primitives).
- It is (supposedly) thread-safe.
- It is (supposedly) a high-performance data structure
package com.fhs.RingBuffer;
import java.lang.reflect.Array;
/**
* Generic, thread-safe, unbounded RingBuffer
*
* @author Ben.Cole
*
* @param <T> Element type
*/
public class RingBuffer<T> {
/** Type of contained Objects */
private Class<T> clazz;
/** Backing array */
private volatile T[] objs;
/** Null array (of size [buffer]) for clearing backing array */
private final T[] nulls;
/** Buffer size - initialized to DEFAULT_BUFFER_SZ */
private int buffer = DEFAULT_BUFFER_SZ;
/** Index of current buffer start */
private volatile int index = 0;
/** Next open spot in backing array */
private volatile int open = 0;
/** Default initial backing array size */
private static final int DEFAULT_ARRAY_SZ = 25;
/** Default buffer size */
private static final int DEFAULT_BUFFER_SZ = 5;
/**
* @param clz Generic Type of this RingBuffer's contents
* @param cap Initial capacity
* @param buff Buffer size
*/
public RingBuffer(Class<T> clz, int cap, int buff) {
this.buffer = buff;
this.clazz = clz;
this.objs = getArray(cap);
this.nulls = getArray(this.buffer);
}
/**
* @param clz Generic Type of this RingBuffer's contents
* @param buff Buffer size
*/
public RingBuffer(Class<T> clz, int buff) {
this(clz, DEFAULT_ARRAY_SZ, buff);
}
/**
* Add an object to this RingBuffer. May result in a resize if load factor after adding element passes the internal limit.
*
* @param obj Object to add to this buffer
*/
public synchronized void add(T obj) {
this.objs[this.open] = obj;
this.open = (this.open + 1) % this.objs.length;
if (shouldExpand()) {
resize();
}
}
/**
* Pop (remove) the next [buffer]'s worth of object from this RingBuffer.
*
* @return An array of T objects, up to [buffer] in size, but could be empty!
*/
public synchronized T[] get() {
T[] retBuff = getArray(this.buffer);
if (this.open < this.index) { // wrapped
int segmentLength = this.objs.length - this.index;
if (segmentLength < this.buffer) { // buffer contents are wrapped
// copy first segment of buffer from backing array to returned array
System.arraycopy(this.objs, this.index, retBuff, 0, segmentLength);
// check to see if we have enough overflow to fill the buffer
int overflow = this.buffer - segmentLength;
boolean fillCheck = this.open >= overflow;
if (fillCheck) { // can completely fill return buffer
// copy overflow segment of buffer from backing array to returned array
System.arraycopy(this.objs, 0, retBuff, segmentLength, overflow);
// copy nulls from null array to backing array
System.arraycopy(this.nulls, 0, this.objs, 0, overflow);
System.arraycopy(this.nulls, 0, this.objs, this.index, segmentLength);
// update index
this.index = overflow;
} else { // can't completely fill return buffer
// copy overflow segment of buffer from backing array to returned array
System.arraycopy(this.objs, 0, retBuff, segmentLength, this.open);
// copy nulls from null array to backing array (same as if we could fill buffer)
System.arraycopy(this.nulls, 0, this.objs, this.index, segmentLength);
// copy nulls to remainder
System.arraycopy(this.nulls, 0, this.objs, 0, this.open);
this.index = this.open;
}
} else { // buffer is *not* actually wrapped!
// copy contents of buffer from backing array to returned array
System.arraycopy(this.objs, this.index, retBuff, 0, this.buffer);
// copy nulls from null array to backing array
System.arraycopy(this.nulls, 0, this.objs, this.index, this.buffer);
// update index, mod-ing by backing array length to account for wrapping
this.index = (this.index + this.buffer) % this.objs.length;
}
} else if (this.index < this.open) { // not wrapped
int between = this.open - this.index;
if (between < this.buffer) { // insufficient elements
// copy contents of buffer from backing array to returned array
System.arraycopy(this.objs, this.index, retBuff, 0, between);
// copy nulls from null array to backing array
System.arraycopy(this.nulls, 0, this.objs, this.index, between);
this.index = this.open;
} else { // sufficient elements
// copy contents of buffer from backing array to returned array
System.arraycopy(this.objs, this.index, retBuff, 0, this.buffer);
// copy nulls from null array to backing array
System.arraycopy(this.nulls, 0, this.objs, this.index, this.buffer);
// update index - don't have to mod the result because
// we already know that we have sufficent space between
// the index and the next open space to fill the buffer
// completely without wrapping around.
this.index = (this.index + this.buffer);
}
} else if (this.index == this.open) {
// return empty buffer
}
return retBuff;
}
/**
* @return true if this buffer is empty.
*/
public boolean isEmpty() {
return this.index == this.open;
}
/**
* @return Number of elements in this RingBuffer
*/
public int size() {
int sz = 0;
if (this.index < this.open) {
sz = this.open - this.index;
} else if (this.index > this.open) {
sz = this.open + (this.objs.length - this.index);
}
return sz;
}
/**
* @return Current size of backing array
*/
protected int backingSize() {
return this.objs.length;
}
/**
* Create a new array of type T using reflection.
*
* @param size Size of desired array
* @return A new array of type T
*/
private T[] getArray(int size) {
return (T[]) Array.newInstance(this.clazz, size);
}
/**
* Checks if the internal load factor (number of filled slots vs number of total slots) has reach a set limit
*
* @return true TODO - SHOULDEXPAND 1 SLOT LEFT
*/
private boolean shouldExpand() {
return
// same as empty check, but since we know that
// we just added an item, we know the buffer isn't
// empty. So it's a 'full?' check.
(this.open == this.index-1)
||
// and check edge case for array wrap
(this.index == 0 && this.open == this.objs.length-1);
}
/**
* Resize the backing array and copy the elements from the old array to the new array, preserving order.
*/
private void resize() {
int newSize = getNewSize();
T[] newObjs = (T[]) Array.newInstance(this.clazz, newSize);
if (this.open < this.index) {
// buffer is wrapped around end of array
int firstSegmentLength = this.objs.length - this.index;
// copy first part of contents
System.arraycopy(this.objs, this.index, newObjs, 0, firstSegmentLength);
// copy second part of contents
System.arraycopy(this.objs, 0, newObjs, firstSegmentLength, this.open);
// update markers
this.index = 0;
this.open = this.open + firstSegmentLength;
} else if (this.index == 0 /* && this.open == this.objs.length - 1 */) {
System.arraycopy(this.objs, 0, newObjs, 0, this.open);
// index and open stay the same
}
this.objs = newObjs;
}
/**
* TODO - RESIZED ARRAY SIZE [Y = X * 2]
*
* @return New backing array size
*/
private int getNewSize() {
int oldSize = this.objs.length;
// double the backing array size
return oldSize * 2;
}
}
Specific questions:
- Is this genuinely thread-safe? I'm pretty sure it is, but I'm not confident enough to say either way.
- Is this truly a high-performance data structure? I believe that it is, based on limited testing (8 producers, 1 consumer, 80k string messages per producer, confident that no messages were dropped).
- Is there a use-case for a data structure of this type? I have no idea on this one - this data structure is something I literally dreamed up and coded the next day. I've never used something like it, but I can imagine it could be useful for burst-tolerant message passing.
