Return to Question

The CircularRing pre-allocates LoggableEntity elements and stores them in an plain array. 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.

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

2. Producer should never blocks 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.

Edit

public final class CircularRing{
    
    private final int ringCapacity;
    private final AtomicInteger writerIndex;
    private final AtomicInteger producerIndex;
    private final LoggableEntity[] ringBuffer;
    
    private final static int DEFAULT_RING_CAPACITY  = SIXTY_FOUR * SIXTY_FOUR;
    private final static Logger LOGGER              = LoggerFactory.getLogger("Ring");

    public CircularRing( ){
        this( DEFAULT_RING_CAPACITY );
    }    
        
    
    public CircularRing( int ringCapacity ){
       
        if( !powerOfTwo( ringCapacity ) ){
            throw new IllegalArgumentException("Ring capacity [" + ringCapacity + "] must be a power of 2.");
        }
        
        this.ringCapacity     = ringCapacity;
        this.writerIndex      = new AtomicInteger( NEGATIVE_ONE );
        this.producerIndex    = new AtomicInteger( NEGATIVE_ONE );
        this.ringBuffer       = new LoggableEntity[ ringCapacity ];
    
        for( int i=ZERO; i< ringCapacity; i++ ){
            ringBuffer[ i ] = new LoggableEntity( );
        }
        
        LOGGER.info("Successfully created Ring with a capacity of {}.", ringCapacity );
    
    }
    
   
    protected final int getWriterCount(){
        return writerIndex.get();
    }
    
    
    protected final LoggableEntity poll( int index ){
        return ringBuffer[ index ];
    }
    
    
    public final int getRingCapacity(){
        return ringCapacity;
    }
    
    
    public final LoggableEntity pollNext(){
        for( ;; ){
            int current = producerIndex.get();
            int next    = current + ONE;
            
            if( producerIndex.compareAndSet(current, next) ){
                next    = (next % ringCapacity);    
                return ringBuffer[ next ];
            }
        }
    }
    
    
    public final void offer( LoggableEntity entity ){
        for( ;; ){
            int current = writerIndex.get();
            int next    = current + ONE;
           
            if( writerIndex.compareAndSet(current, next) ){
                next    = (next % ringCapacity);
                ringBuffer[ next ]  = entity;
                return;
            }
        }
    }
       
}

public final class RingLogger{
        
        private volatile boolean keepLogging;
        
        private final CircularRing eRing;
        private final BackgroundLogger backLogger;
        private final ExecutorService executor;
            
        private final static int DEFAULT_BLOCK_SIZE = THIRTY_TWO * SIXTY_FOUR;
        private final static Logger LOGGER          = LoggerFactory.getLogger( "RingLogger" ); 
        
        
        public RingLogger( int ringSize, LogWriter writer ) throws IOException{
            this( DEFAULT_BLOCK_SIZE, ringSize, writer, new DefaultMessageFormatter() );
        }
        
        public RingLogger( int blockSize, int ringSize, LogWriter writer ) throws IOException{
            this( blockSize, ringSize, writer, new DefaultMessageFormatter() );
        }
        
        
        public RingLogger( int blockSize, int ringSize, LogWriter writer, MessageFormatter formatter ) throws IOException{
            this.eRing          = new CircularRing( ringSize );
            this.backLogger     = new BackgroundLogger( formatter, writer );
            this.executor       = Executors.newCachedThreadPool( );
        }
        
        
        public final void init(){
            keepLogging = true;
            executor.execute( backLogger );
            
            LOGGER.info("Successfully initialized logger.");
        }
            
        
        public final LoggableEntity pollNext( ){
            return eRing.pollNext( );
        }
        
                
        public final void offer( LoggableEntity data ){
            eRing.offer( data );
        }
            
       
        public final void stop(){
            try {
                LOGGER.info("Received request to stop.");
                executor.shutdown();
                executor.awaitTermination( TWO, TimeUnit.SECONDS );
                keepLogging = false;
                
            }catch( Exception e ){
                LOGGER.warn("Exception while stopping logger.", e);
            }
        
        }
       
    
            
            private final class BackgroundLogger implements Runnable{
        
                private int ringWriterCount     = NEGATIVE_ONE;
                private int logWriterCount      = NEGATIVE_ONE;
                
                private final int blockSize;
                private final int ringCapacity;
                private final LogWriter writer;
                private final MessageFormatter formatter;
                private final ReusableStringBuilder builder;
                
                public BackgroundLogger( MessageFormatter formatter, LogWriter writer ){
                    this.writer         = writer;
                    this.formatter      = formatter;
                    this.blockSize      = writer.getBlockSize();
                    this.builder        = new ReusableStringBuilder();
                    this.ringCapacity   = eRing.getRingCapacity();
                }
                
                
                public void run(){
                    
                    while( keepLogging ){
        
                        while ( keepLogging && ( logWriterCount == (ringWriterCount = circularRing.getWriterCount()) ) ){
                            LockSupport.parkNanos( ONE );
                        }
                    
                        int entityCount = 0;
                              
                        while( (entityCount < blockSize) && (logWriterCount < ringWriterCount) ){
                            
                            int writerIndex = logWriterCount + ONE;
                            writerIndex     = (writerIndex >= ringCapacity) ? (writerIndex % ringCapacity) : writerIndex;
                   
                            LoggableEntity data = circularRing.poll( writerIndex );
                            formatter.formatMessage( builder, data );
                        
                            ++logWriterCount;
                            ++entityCount;
                        }
                    
                        writer.writeBuffered( builder );
                        builder.wipeout();
                    
                    }
                
                    LOGGER.info("Background Logger thread successfully stopped.");
                }
                
            }
                
        }

public final class LoggableEntity{
        
            private Level level;
            private String msg;
            
            public void debug( String msg ){
                populateLog( DEBUG, name );
            }
            
            public void info( String name, String msg ){
                populateLog( INFO, name );
            }
        
            private final void populateLog( Level level, String msg ){
                this.level     = level;
                this.msg       = msg;
            }
        
           public final Level getLevel( ){
                return level;
            }
            
            
            public final String getMessage( ){
                return msg;
            }
        
        }

public class Tester{


    public static void main( String[] args ) throws Exception{

            int iteration       = 16;
            int blockSize       = 8;
        String TEMPLATE     = "This is a simulated dummy log message number ";
          
            LogWriter writer    = new DefaultLogWriter( blockSize, "Test.log");
            RingLogger logger   = new RingLogger( iteration, writer );
            logger.init();
            
            for( int i =0; i < iteration; i++ ){
               LoggableEntity data = logger.pollNext( );
               data.debug( NAME, TEMPLATE + i );
               logger.offer( data );
            }
     
            logger.stop();

  }

}

Any comments, pointers etc are appreciated. Thank you and happy holidays.

-Cheers

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.

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.

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.

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 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.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 consumerIndex.incrementAndGet();
}

//Unsued
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 producerIndex.incrementAndGet();
}


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


public final void addLazily( int index, LoggableEntity entity ){
    ringBuffer.lazySet( (index % ringCapacity), 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 );
                
    }

    
}

The CircularRing pre-allocates LoggableEntity elements and stores them in an plain array. 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.

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

2. Producer should never blocks 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.

Edit

public final class CircularRing{
    
    private final int ringCapacity;
    private final AtomicInteger writerIndex;
    private final AtomicInteger producerIndex;
    private final LoggableEntity[] ringBuffer;
    
    private final static int DEFAULT_RING_CAPACITY  = SIXTY_FOUR * SIXTY_FOUR;
    private final static Logger LOGGER              = LoggerFactory.getLogger("Ring");

    public CircularRing( ){
        this( DEFAULT_RING_CAPACITY );
    }    
        
    
    public CircularRing( int ringCapacity ){
       
        if( !powerOfTwo( ringCapacity ) ){
            throw new IllegalArgumentException("Ring capacity [" + ringCapacity + "] must be a power of 2.");
        }
        
        this.ringCapacity     = ringCapacity;
        this.writerIndex      = new AtomicInteger( NEGATIVE_ONE );
        this.producerIndex    = new AtomicInteger( NEGATIVE_ONE );
        this.ringBuffer       = new LoggableEntity[ ringCapacity ];
    
        for( int i=ZERO; i< ringCapacity; i++ ){
            ringBuffer[ i ] = new LoggableEntity( );
        }
        
        LOGGER.info("Successfully created Ring with a capacity of {}.", ringCapacity );
    
    }
    
   
    protected final int getWriterCount(){
        return writerIndex.get();
    }
    
    
    protected final LoggableEntity poll( int index ){
        return ringBuffer[ index ];
    }
     
    
    public final int getRingCapacity(){
        return ringCapacity;
    }
    
    
    public final LoggableEntity pollNext(){
        for( ;; ){
            int current = producerIndex.get();
            int next    = current + ONE;
            
            if( producerIndex.compareAndSet(current, next) ){
                next    = (next % ringCapacity);    
                return ringBuffer[ next ];
            }
        }
    }
    
    
    public final void offer( LoggableEntity entity ){
        for( ;; ){
            int current = writerIndex.get();
            int next    = current + ONE;
           
            if( writerIndex.compareAndSet(current, next) ){
                next    = (next % ringCapacity);
                ringBuffer[ next ]  = entity;
                return;
            }
        }
    }
       
}

public final class RingLogger{
        
        private volatile boolean keepLogging;
        
        private final CircularRing eRing;
        private final BackgroundLogger backLogger;
        private final ExecutorService executor;
            
        private final static int DEFAULT_BLOCK_SIZE = THIRTY_TWO * SIXTY_FOUR;
        private final static Logger LOGGER          = LoggerFactory.getLogger( "RingLogger" ); 
        
        
        public RingLogger( int ringSize, LogWriter writer ) throws IOException{
            this( DEFAULT_BLOCK_SIZE, ringSize, writer, new DefaultMessageFormatter() );
        }
        
        public RingLogger( int blockSize, int ringSize, LogWriter writer ) throws IOException{
            this( blockSize, ringSize, writer, new DefaultMessageFormatter() );
        }
        
        
        public RingLogger( int blockSize, int ringSize, LogWriter writer, MessageFormatter formatter ) throws IOException{
            this.eRing          = new CircularRing( ringSize );
            this.backLogger     = new BackgroundLogger( formatter, writer );
            this.executor       = Executors.newCachedThreadPool( );
        }
         
        
        public final void init(){
            keepLogging = true;
            executor.execute( backLogger );
            
            LOGGER.info("Successfully initialized logger.");
        }
            
        
        public final LoggableEntity pollNext( ){
            return eRing.pollNext( );
        }
        
                
        public final void offer( LoggableEntity data ){
            eRing.offer( data );
        }
            
       
        public final void stop(){
            try {
                LOGGER.info("Received request to stop.");
                executor.shutdown();
                executor.awaitTermination( TWO, TimeUnit.SECONDS );
                keepLogging = false;
                
            }catch( Exception e ){
                LOGGER.warn("Exception while stopping logger.", e);
            }
        
        }
       
    
            
            private final class BackgroundLogger implements Runnable{
        
                private int ringWriterCount     = NEGATIVE_ONE;
                private int logWriterCount      = NEGATIVE_ONE;
                
                private final int blockSize;
                private final int ringCapacity;
                private final LogWriter writer;
                private final MessageFormatter formatter;
                private final ReusableStringBuilder builder;
                
                public BackgroundLogger( MessageFormatter formatter, LogWriter writer ){
                    this.writer         = writer;
                    this.formatter      = formatter;
                    this.blockSize      = writer.getBlockSize();
                    this.builder        = new ReusableStringBuilder();
                    this.ringCapacity   = eRing.getRingCapacity();
                }
                
                
                public void run(){
                    
                    while( keepLogging ){
        
                        while ( keepLogging && ( logWriterCount == (ringWriterCount = circularRing.getWriterCount()) ) ){
                            LockSupport.parkNanos( ONE );
                        }
                    
                        int entityCount = 0;
                              
                        while( (entityCount < blockSize) && (logWriterCount < ringWriterCount) ){
                            
                            int writerIndex = logWriterCount + ONE;
                            writerIndex     = (writerIndex >= ringCapacity) ? (writerIndex % ringCapacity) : writerIndex;
                    
                            LoggableEntity data = circularRing.poll( writerIndex );
                            formatter.formatMessage( builder, data );
                        
                            ++logWriterCount;
                            ++entityCount;
                        }
                    
                        writer.writeBuffered( builder );
                        builder.wipeout();
                    
                    }
                
                    LOGGER.info("Background Logger thread successfully stopped.");
                }
                
            }
                
        }

public final class LoggableEntity{
        
            private Level level;
            private String msg;
            
            public void debug( String msg ){
                populateLog( DEBUG, name );
            }
            
            public void info( String name, String msg ){
                populateLog( INFO, name );
            }
         
            private final void populateLog( Level level, String msg ){
                this.level     = level;
                this.msg       = msg;
            }
        
           public final Level getLevel( ){
                return level;
            }
            
            
            public final String getMessage( ){
                return msg;
            }
        
        }

public class Tester{


    public static void main( String[] args ) throws Exception{

            int iteration       = 16;
            int blockSize       = 8;
        String TEMPLATE     = "This is a simulated dummy log message number ";
          
            LogWriter writer    = new DefaultLogWriter( blockSize, "Test.log");
            RingLogger logger   = new RingLogger( iteration, writer );
            logger.init();
            
            for( int i =0; i < iteration; i++ ){
               LoggableEntity data = logger.pollNext( );
               data.debug( NAME, TEMPLATE + i );
               logger.offer( data );
            }
     
            logger.stop();

  }

}