Added support for timeouts on consuming from the ring buffer

Author: mikeb01

README.md

@@ -11,6 +11,13 @@ Maintainer
 Changelog
 ==========
 
+## 3.0.0.beta3 Released (20-Feb-2013)
+
+- Significant Javadoc updates (thanks Jason Koch)
+- DSL support for WorkerPool
+- Small performance tweaks
+- Add TimeoutHandler and TimeoutBlockingWaitStrategy and support timeouts in BatchEventProcessor
+
 ## 3.0.0.beta2 Released (7-Jan-2013)
 
 - Remove millisecond wakeup from BlockingWaitStrategy

build.gradle

@@ -22,7 +22,7 @@ apply plugin: 'eclipse'   // Only used so the Eclipse STS Gradle plugin can see
 defaultTasks 'build'
 
 group = 'com.lmax'
-version = new Version(major: 3, stage: 'beta2')
+version = new Version(major: 3, stage: 'beta3')
 
 ext {
     fullName = 'Disruptor Framework'

src/main/java/com/lmax/disruptor/BatchEventProcessor.java

@@ -17,6 +17,7 @@
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
+
 /**
  * Convenience class for handling the batching semantics of consuming entries from a {@link RingBuffer}
  * and delegating the available events to an {@link EventHandler}.
@@ -35,6 +36,7 @@ public final class BatchEventProcessor<T>
     private final SequenceBarrier sequenceBarrier;
     private final EventHandler<T> eventHandler;
     private final Sequence sequence = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
+    private final TimeoutHandler timeoutHandler;
 
     /**
      * Construct a {@link EventProcessor} that will automatically track the progress by updating its sequence when
@@ -56,6 +58,8 @@ public BatchEventProcessor(final RingBuffer<T> ringBuffer,
         {
             ((SequenceReportingEventHandler<?>)eventHandler).setSequenceCallback(sequence);
         }
+        
+        timeoutHandler = (eventHandler instanceof TimeoutHandler) ? (TimeoutHandler) eventHandler : null;
     }
 
     @Override
@@ -109,6 +113,13 @@ public void run()
             try
             {
                 final long availableSequence = sequenceBarrier.waitFor(nextSequence);
+                
+                if (availableSequence < nextSequence && timeoutHandler != null)
+                {
+                    notifyTimeout(availableSequence);
+                    continue;
+                }
+                
                 while (nextSequence <= availableSequence)
                 {
                     event = ringBuffer.getPublished(nextSequence);
@@ -138,6 +149,18 @@ public void run()
         running.set(false);
     }
 
+    private void notifyTimeout(final long availableSequence) throws Exception
+    {
+        try
+        {
+            timeoutHandler.onTimeout(availableSequence);
+        }
+        catch (Throwable e)
+        {
+            exceptionHandler.handleEventException(e, availableSequence, null);
+        }
+    }
+
     /**
      * Notifies the EventHandler when this processor is starting up
      */

src/main/java/com/lmax/disruptor/ExceptionHandler.java

@@ -28,7 +28,7 @@ public interface ExceptionHandler
      *
      * @param ex the exception that propagated from the {@link EventHandler}.
      * @param sequence of the event which cause the exception.
-     * @param event being processed when the exception occurred.
+     * @param event being processed when the exception occurred.  This can be null.
      */
     void handleEventException(Throwable ex, long sequence, Object event);
 

src/main/java/com/lmax/disruptor/TimeoutBlockingWaitStrategy.java

@@ -0,0 +1,77 @@
+package com.lmax.disruptor;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import com.lmax.disruptor.AlertException;
+import com.lmax.disruptor.Sequence;
+import com.lmax.disruptor.SequenceBarrier;
+import com.lmax.disruptor.WaitStrategy;
+
+public class TimeoutBlockingWaitStrategy implements WaitStrategy
+{
+    private final Lock lock = new ReentrantLock();
+    private final Condition processorNotifyCondition = lock.newCondition();
+    private final long timeoutInNanos;
+    
+    public TimeoutBlockingWaitStrategy(final long timeout, final TimeUnit units)
+    {
+        timeoutInNanos = units.toNanos(timeout);
+    }
+
+    @Override
+    public long waitFor(final long sequence, 
+                        final Sequence cursorSequence, 
+                        final Sequence dependentSequence, 
+                        final SequenceBarrier barrier)
+        throws AlertException, InterruptedException
+    {
+        long nanos = timeoutInNanos;
+        
+        long availableSequence;
+        if ((availableSequence = cursorSequence.get()) < sequence)
+        {
+            lock.lock();
+            try
+            {
+                while ((availableSequence = cursorSequence.get()) < sequence)
+                {
+                    barrier.checkAlert();
+                    nanos = processorNotifyCondition.awaitNanos(nanos);
+                    if (nanos <= 0)
+                    {
+                        return availableSequence;
+                    }
+                }
+            }
+            finally
+            {
+                lock.unlock();
+            }
+        }
+
+        while ((availableSequence = dependentSequence.get()) < sequence)
+        {
+            barrier.checkAlert();
+        }
+
+        return availableSequence;
+    }
+
+    @Override
+    public void signalAllWhenBlocking()
+    {
+        lock.lock();
+        try
+        {
+            processorNotifyCondition.signalAll();
+        }
+        finally
+        {
+            lock.unlock();
+        }
+    }
+
+}

src/main/java/com/lmax/disruptor/TimeoutHandler.java

@@ -0,0 +1,6 @@
+package com.lmax.disruptor;
+
+public interface TimeoutHandler
+{
+    void onTimeout(long sequence) throws Exception;
+}

src/main/java/com/lmax/disruptor/WaitStrategy.java

@@ -22,7 +22,11 @@
 public interface WaitStrategy
 {
     /**
-     * Wait for the given sequence to be available
+     * Wait for the given sequence to be available.  It is possible for this method to return a value
+     * less than the sequence number supplied depending on the implementation of the WaitStrategy.  A common
+     * use for this is to signal a timeout.  Any EventProcessor that is using a WaitStragegy to get notifications
+     * about message becoming available should remember to handle this case.  The {@link BatchEventProcessor} explicitly
+     * handles this case and will signal a timeout if required. 
      *
      * @param sequence to be waited on.
      * @param cursor the main sequence from ringbuffer. Wait/notify strategies will

src/test/java/com/lmax/disruptor/TimeoutBlockingWaitStrategyTest.java

@@ -0,0 +1,51 @@
+package com.lmax.disruptor;
+
+import java.util.concurrent.TimeUnit;
+
+import com.lmax.disruptor.Sequence;
+import com.lmax.disruptor.SequenceBarrier;
+
+import org.jmock.Expectations;
+import org.jmock.Mockery;
+import org.jmock.integration.junit4.JMock;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.Assert.assertThat;
+
+@RunWith(JMock.class)
+public class TimeoutBlockingWaitStrategyTest
+{
+    private final Mockery mockery = new Mockery();
+    
+    @Test
+    public void shouldTimeoutWaitFor() throws Exception
+    {
+        final SequenceBarrier sequenceBarrier = mockery.mock(SequenceBarrier.class);
+        
+        long theTimeout = 500;
+        TimeoutBlockingWaitStrategy waitStrategy = new TimeoutBlockingWaitStrategy(theTimeout, TimeUnit.MILLISECONDS);
+        Sequence cursor = new Sequence(5);
+        Sequence dependent = cursor;
+        
+        mockery.checking(new Expectations()
+        {
+            {
+                allowing(sequenceBarrier).checkAlert();
+            }
+        });
+        
+        long t0 = System.currentTimeMillis();
+        
+        long sequence = waitStrategy.waitFor(6, cursor, dependent, sequenceBarrier);
+        
+        long t1 = System.currentTimeMillis();
+        
+        long timeWaiting = t1 - t0;
+        
+        assertThat(sequence, is(5L));
+        assertThat(timeWaiting, greaterThanOrEqualTo(theTimeout));
+    }
+}