3
\$\begingroup\$

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:

  1. Is this genuinely thread-safe? I'm pretty sure it is, but I'm not confident enough to say either way.
  2. 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).
  3. 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.
\$\endgroup\$
1
\$\begingroup\$

0) It seems simpler to me to just copy the remainder of the buffer in a normal array every time some chunk is removed instead of using a circular buffer. Using a LinkedList would not even require copying anything. But I must admit if you have millions of bytes, a LinkedList<Byte> would not be a very good idea.

1) I don't like the use of raw arrays and of reflection. However, since this is supposed to be for high-performance, maybe it is alright.

2) I looked at the code lightly and I believe it is correctly synchronized, except that isEmpty() and size() should probably be synchronized too. However, I would not say it is high performance because you are simply locking add() and get(), which is quite crude. You probably don't need to stop a thread from writing to the RingBuffer when another thread is reading it since they likely affect different parts of the buffer. There are other concepts I would explore for high-performance multi-threading: compare-and-swap and copy-on-write. (If you use a LinkedList as I suggested above, I can see that you could hold the lock for only very short periods of time.)

\$\endgroup\$
  • \$\begingroup\$ 0) Not sure what you mean in the first sentence here. I can understand how a linked list would remove the need for copying, though. 1) Same here, I use lists, etc, in more 'regular' code. \$\endgroup\$ – BenCole Sep 17 '13 at 3:48
  • \$\begingroup\$ 2) I thought locking get/add was crude also. I had wondered if, given the possibility of a read/write collision, locking on add/get would be easier (and result in a small enough performance effect) than attempting to allow truly asynchronous access by employing more complex multiplexing strategies. The number of actual operations in the add/get methods is also relatively small (even if a resize occurs, which has a similar footprint to a get() call). However, I will look up compare-and-swap and copy-on-write. \$\endgroup\$ – BenCole Sep 17 '13 at 3:49
0
\$\begingroup\$

Comments for maintenance, simplification, and extensibility:

  1. Consider a static import for System.arraycopy.
  2. Eliminate the magic number and oldSize variable when resizing:
    return this.objs.length * getExpansionSize();
    
  3. Nearly all usages of this. are superfluous.
  4. Class-scoped integer variables are initialized to 0 by default.
  5. Stylistically, declare the constants before variables (e.g., define DEFAULT_BUFFER_SZ before use).
  6. The fillcheck variable is not required (compiler should optimize it away).
  7. While returning from a method at a single location is desirable, the size() method could be shortened by returning multiple times, eliminating the sz variable and corresponding assignment to 0 (compiler might do this?).
  8. The following line:
    T[] newObjs =  (T[]) Array.newInstance(this.clazz, newSize);
    
    can be:
    T[] newObjs = getArray(getNewSize());
    
    to remove duplication and eliminate a variable.
  9. Minor (insignificant) optimization, in the constructor change:
    this.nulls = getArray(this.buffer);
    to:
    this.nulls = getArray(buff)
    as referencing a local variable uses a shorter bytecode than referencing a class-scoped variable.
  10. Remove the following code:
    else if (this.index == this.open) {
    // return empty buffer
    }
    by commenting the return statement as:
    // retBuff will be empty if index == open.

I, personally, prefer result over retBuff; return statement is read as "return the result" vs. "return the return buffer." Throughout my code, the result variable, invariably, is the variable that contains the value returned from a function.

For extensibility, a private getArray(int) method prevents subclasses from adding injecting custom Array behaviour. Also, a get method name prefix is typically reserved for accessing a member attribute; a create method name prefix is typical for methods that simply perform instantiation. Thus you could have a protected method called createArray(int) and another helper method of protected createArray(Class<T>, int) that subclasses could override.

\$\endgroup\$
0
\$\begingroup\$

0) - agreed with toto2

2) - no. You did not say how many messages per second your buffer can pass through itself (and did not provide test code to run), but high performance counts in millions, and I doubt your buffer can give even 1 M mps. Main performance loss is copying data in get(). It takes time to allocate, copy, and deallocate arrays. The latter is the worst, as it causes unpredictable garbage collector executions which dramatically increase reaction time.

3) look at Disruptor - the most performant ring buffer to connect producer and consumer threads (see RingBuffer class source code).

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.