Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.

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.

share|improve this question
    
Why are you exposing the index? The very fact that it's a ring buffer is immaterial; you should implement java.util.Queue or a sub-interface (BlockingQueue? assuming this is for learning, otherwise you should probably be using an existing implementation). final classes don't need methods marked final. Don't instantiate other classes in constructors, inject them instead. LoggableEntity should be immutable - your current implementation is not threadsafe. –  Clockwork-Muse Dec 27 '12 at 18:43
    
Thanks but this Ring impl behaves differently from a Queue/BlockingQueue. The ring pre-populates LoggableEntity objects. Poll() returns the object and add() stores it back into the ring. Also, the producer never blocks. I thought of using ConcurrentLinkedQueue but it's quite heavy and I don't need its additional features. The indices are exposed to ensure the atomicity of offers()/puts(). If either of them first look up an index and then stores/removes objects, the method would have to be locked. incrementModAndGet() is indeed broken and I am hoping to find a way to fix it without locking. –  CaptainHastings Dec 28 '12 at 20:48
    
Pre-populating your LoggableEntity objects is actually a problem (because they aren't threadsafe). You conceptually have a queue, the fact that internally its implemented with a buffer is immaterial; personally, I'd probably use an ArrayBlockingQueue. Currently, with the producer not blocking, you're overwriting elements that potentially haven't been consumed yet. If the producer shuts down, you will consume elements multiple times. Exposing the indices has no effect on atomicity. –  Clockwork-Muse Dec 28 '12 at 21:19
    
You may find this of interest. It is not what you want but it may be a better starting point. –  OldCurmudgeon Dec 29 '12 at 1:40
    
Thanks @OldCurmudgeon, I will have a look. –  CaptainHastings Jan 3 '13 at 18:29
show 1 more comment

2 Answers 2

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/

share|improve this answer
    
Thanks Christian. I do want to let the producers wrap around the ring and overwrite the older entries. Can you please explain how that would make it not thread-safe? Point well taken about bulk reads. I plan to have just one consumer reading from the buffer and on a second look, consumerIndex variable can just be a volatile int (as opposed to AtomicInteger) as it will only be incremented by one consumer thread. Funny that you mention Disruptor. My ring idea is honorably borrowed from Disruptor :) but I wanted to write my own implementation. –  CaptainHastings Jan 3 '13 at 18:27
    
Wrapping around isn't a problem as long as you somehow ensure that you don't overwrite entries that have been consumed. As I read the code (and I could be wrong) it looks like the producer can run ahead of the consumer, overwriting entries that either have not been consumed or are in the process of being consumed, and that's a problem when they are mutable (and mutable entries are part of the point of these kinds of ring-buffers - important for performance by reducing the allocation rate). –  Chris Vest Jan 7 '13 at 16:07
    
Hi Christian, thanks once again. I have modified the code to include, what disruptor calls a two-phase commit. In phase 1, producer looks up the entity from the next index in the array and modifies it. In phase 2, it stores it back in the array. The background logger now tracks the phase-2 index. So the background thread will only writes those entities which have been modified and stored. –  CaptainHastings Jan 8 '13 at 15:32
add comment

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.

share|improve this answer
    
Thanks Andras. I like your idea of modding when inserting into the ring. I checked and looks like the mod function isn't thread safe either. However, in spite of it, I think your solution will work. –  CaptainHastings Dec 28 '12 at 20:22
    
To elaborate, since the mod (%) function itself isn't thread safe, I am unsure how addLazily() will behave if multiple threads calls it at the same time. To be honest, my implementation of incrementModAndGet() is broken as well. The else part is atomic where as the if part isn't. I can't think of any other way to make it thread-safe without locking it (which defeats the whole purpose of attempting to write a lock-free impl). Any additional ideas would be most welcome. Thanks. –  CaptainHastings Dec 28 '12 at 20:29
add comment

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

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