Lock-Free Ring Implementation

Question

I am wondering if someone can take a look at my lock-free, circular ring implementation which I use to implement a background logger.

The CircularRing pre-allocates LoggableEntity elements and stores them in an AtomicReferenceArray. Then, I have multiple producer threads which will enqueue elements and single consumer thread which will dequeue elements in a bulk and write them to a storage device.

I track the index where the next produced item and the next consumed item should come from. Once the capacity is reached, the indices wrap.

My goals behind this implementations are the following:

1. Create a lock free, thread-safe data structure which can store pre-allocated elements.

2. Producer should never wait to enqueue but rather wrap if the capacity is reached.

3. The relative order of elements within a producer (thread) must be respected. For example, if producer1 and producer2 produces A, B, C and P, Q, R at the same time; any interspersing of the elements is fine as long as the relative order is maintained.

4. Finally, when the consumer index is equal to the producer index; the consumer waits by busy spinning.

public final class LoggableEntity {

    private long time;
    private String message;

    public static final LoggableEntity getInstance() {
        return new LoggableEntity();
    }

    //Setters and Getters
}


public final class CircularRing {

    private final int ringCapacity;
    private final int ringCapacityIndex;
    private final AtomicInteger consumerIndex;
    private final AtomicInteger producerIndex;
    private final AtomicReferenceArray<LoggableEntity> ringBuffer;

    private final static int DEFAULT_BUFFER_CAPACITY    = 1024;
    private final static int DEFAULT_RING_CAPACITY      = 16 * 1024;

    public CircularRing( ) {
        this( DEFAULT_RING_CAPACITY, DEFAULT_BUFFER_CAPACITY );
}    


public CircularRing( int ringCapacity, int bufferCapacity ) {
    this.ringCapacity       = ringCapacity;
    this.ringCapacityIndex  = ringCapacity -1;
    this.consumerIndex      = new AtomicInteger( -1 );
    this.producerIndex      = new AtomicInteger( -1 );
    this.ringBuffer         = new AtomicReferenceArray<LoggableEntity>( ringCapacity );

    for( int i =0; i< ringCapacity; i++ ) {
        addLazily( i , LoggableEntity.getInstance() );
    }
}

public final int getCurrentConsumerIndex() {
    return consumerIndex.get();
}


public final int getNextConsumerIndex() {
    return incrementModAndGet( consumerIndex );
}


private final int incrementModAndGet( AtomicInteger aInt ) {

    if ( aInt.get() < ringCapacityIndex ) {
        return aInt.incrementAndGet();

    } else {

        for (;;) {
            int current = aInt.get();
            int next    = (current + 1) % ringCapacity;
            if( aInt.compareAndSet(current, next) )
            return next;
        }
    }
}


public final int getCurrentProducerIndex() {
    return producerIndex.get();
}


public final int getNextProducerIndex() {
    return incrementModAndGet( producerIndex );
}


public final LoggableEntity poll( int index ) {
    return ringBuffer.get( index );
}


public final void addLazily( int index, LoggableEntity entity ) {
    ringBuffer.lazySet( index, entity );
}

Logger

public final class CircularSmartLogger {

    private volatile boolean keepLogging;

    private final CircularRing circularRing;
    private final ExecutorService executor;
    private final BackgroundCircularLogger backLogger;

    public CircularSmartLogger( int bulkSize ){
        this.circularRing   = new CircularRing();
        this.backLogger     = new BackgroundCircularLogger( bulkSize );
        this.executor       = Executors.newCachedThreadPool();
    }


    public final void init() {
        keepLogging = true;
        executor.execute( backLogger );
    }


    public final int getNextProducerIndex( ) {
        return circularRing.getNextProducerIndex();
    }

    public final LoggableEntity poll( int index ) {
        return circularRing.poll( index );
    }

    public final void addLazily( int index, LoggableEntity data ) {
        circularRing.addLazily( index, data );
    }


    public final void stop() {
        keepLogging = false;
    }


    private final class BackgroundCircularLogger implements Runnable {

        private int pIndex;
        private int cIndex;
        private final int bulkSize;

        public BackgroundCircularLogger( int bulkSize ) {
            this.pIndex     = -1;
            this.cIndex     = -1;
            this.bulkSize   = bulkSize;
        }

        @Override
        public void run( ) {

            while( keepLogging ) {

                while ( (cIndex = circularRing.getCurrentConsumerIndex()) == (pIndex = circularRing.getCurrentProducerIndex()) ) {
                    LockSupport.parkNanos( 1L );
                }

                int items    = pIndex - cIndex;
                items        = ( items > bulkSize ) ? bulkSize : items;

                for ( int i =0; i < items; i++ ) {
                    int nextIdx             = circularRing.getNextConsumerIndex();
                    LoggableEntity data     = circularRing.poll( nextIdx );

                    //Write attributes of "data" it to a Storage Device
                }                  
            }
        }   
    } 
}

Tester

public class CircularRingTester {

    public static void main( String[] args ) {

        CircularSmartLogger logger = new CircularSmartLogger( 50 );
        logger.init();

        int next            = logger.getNextProducerIndex();
        LoggableEntity data = logger.poll( next );

        data.setTime( System.currentTimeMillis() );
        data.setMessage("This is a simulated message.");

        logger.addLazily( next, data );  
    }
}

Any comments, pointers etc are appreciated.

Answer 1

If the producers run faster than the consumer, then entries in the CircularRing will eventually be overwritten. This will also expose concurrent access to the LoggableEntity objects, which isn't thread-safe.

When you are doing bulk-reads from the buffer, you can also bulk-increment the consumer index and potentially save CAS instructions.

If this buffer isn't explicitly single-consumer, then the bulk-read code is racy and can run ahead of the producers. Again, potentially causing concurrent access to the LoggableEntities.

Have you considered using the Disruptor library? http://lmax-exchange.github.com/disruptor/

Answer 2

I think there is a synchronization problem in incrementModAndGet. It is not synchronized, thus it can happen that two threads reaches the first if when the current index is at the end of the buffer, both passes the condition and the second that calls to aInt.incrementAndGet() will get an index that is outside of the valid range.

Instead of trying to find a suitable index with mod, why don't you go with one sequence of numbers and use mod when you insert into the ring? I.e. in CircularRing

public final void addLazily( int index, LoggableEntity entity ){
    ringBuffer.lazySet( index % ringCapacity, entity );
}

Also you could sophisticate the tester class somewhat more, by using more threads.