Ensure that if halt is call the BatchEventProcessor will always exit …

…cleanly. #111.

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

@@ -15,7 +15,7 @@
  */
 package com.lmax.disruptor;
 
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 
 /**
@@ -30,7 +30,11 @@
 public final class BatchEventProcessor<T>
     implements EventProcessor
 {
-    private final AtomicBoolean running = new AtomicBoolean(false);
+    private static final int IDLE = 0;
+    private static final int HALTED = IDLE + 1;
+    private static final int RUNNING = HALTED + 1;
+
+    private final AtomicInteger running = new AtomicInteger(IDLE);
     private ExceptionHandler<? super T> exceptionHandler = new FatalExceptionHandler();
     private final DataProvider<T> dataProvider;
     private final SequenceBarrier sequenceBarrier;
@@ -76,14 +80,14 @@ public Sequence getSequence()
     @Override
     public void halt()
     {
-        running.set(false);
+        running.set(HALTED);
         sequenceBarrier.alert();
     }
 
     @Override
     public boolean isRunning()
     {
-        return running.get();
+        return running.get() != IDLE;
     }
 
     /**
@@ -109,18 +113,29 @@ public void setExceptionHandler(final ExceptionHandler<? super T> exceptionHandl
     @Override
     public void run()
     {
-        if (!running.compareAndSet(false, true))
+        if (!running.compareAndSet(IDLE, RUNNING))
         {
-            throw new IllegalStateException("Thread is already running");
+            if (running.get() == RUNNING)
+            {
+                throw new IllegalStateException("Thread is already running");
+            }
         }
+        // halt();
         sequenceBarrier.clearAlert();
 
         notifyStart();
 
-        T event = null;
-        long nextSequence = sequence.get() + 1L;
         try
         {
+            if (running.get() == HALTED)
+            {
+                running.set(IDLE);
+                return;
+            }
+
+            T event = null;
+            long nextSequence = sequence.get() + 1L;
+
             while (true)
             {
                 try
@@ -146,7 +161,7 @@ public void run()
                 }
                 catch (final AlertException ex)
                 {
-                    if (!running.get())
+                    if (running.get() != RUNNING)
                     {
                         break;
                     }
@@ -162,7 +177,7 @@ public void run()
         finally
         {
             notifyShutdown();
-            running.set(false);
+            running.set(IDLE);
         }
     }
 

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

@@ -187,4 +187,91 @@ public void onEvent(StubEvent event, long sequence, boolean endOfBatch)
 
         assertEquals(Arrays.asList(3L, 2L, 1L), batchSizes);
     }
+
+    @Test
+    public void shouldAlwaysHalt() throws InterruptedException
+    {
+        WaitStrategy waitStrategy = new BusySpinWaitStrategy();
+        final SingleProducerSequencer sequencer = new SingleProducerSequencer(8, waitStrategy);
+        final ProcessingSequenceBarrier barrier = new ProcessingSequenceBarrier(
+            sequencer, waitStrategy, new Sequence(-1), new Sequence[0]);
+        DataProvider<Object> dp = new DataProvider<Object>()
+        {
+            @Override
+            public Object get(long sequence)
+            {
+                return null;
+            }
+        };
+
+        final LatchLifeCycleHandler h1 = new LatchLifeCycleHandler();
+        final BatchEventProcessor p1 = new BatchEventProcessor<>(dp, barrier, h1);
+
+        Thread t1 = new Thread(p1);
+        p1.halt();
+        t1.start();
+
+        assertTrue(h1.awaitStart(2, TimeUnit.SECONDS));
+        assertTrue(h1.awaitStop(2, TimeUnit.SECONDS));
+
+        for (int i = 0; i < 1000; i++)
+        {
+            final LatchLifeCycleHandler h2 = new LatchLifeCycleHandler();
+            final BatchEventProcessor p2 = new BatchEventProcessor<>(dp, barrier, h2);
+            Thread t2 = new Thread(p2);
+            t2.start();
+            p2.halt();
+
+            assertTrue(h2.awaitStart(2, TimeUnit.SECONDS));
+            assertTrue(h2.awaitStop(2, TimeUnit.SECONDS));
+        }
+
+        for (int i = 0; i < 1000; i++)
+        {
+            final LatchLifeCycleHandler h2 = new LatchLifeCycleHandler();
+            final BatchEventProcessor p2 = new BatchEventProcessor<>(dp, barrier, h2);
+            Thread t2 = new Thread(p2);
+            t2.start();
+            Thread.yield();
+            p2.halt();
+
+            assertTrue(h2.awaitStart(2, TimeUnit.SECONDS));
+            assertTrue(h2.awaitStop(2, TimeUnit.SECONDS));
+        }
+    }
+
+    private static class LatchLifeCycleHandler implements EventHandler<Object>, LifecycleAware
+    {
+        private final CountDownLatch startLatch = new CountDownLatch(1);
+        private final CountDownLatch stopLatch = new CountDownLatch(1);
+
+        @Override
+        public void onEvent(Object event, long sequence, boolean endOfBatch) throws Exception
+        {
+
+        }
+
+        @Override
+        public void onStart()
+        {
+            startLatch.countDown();
+        }
+
+        @Override
+        public void onShutdown()
+        {
+            stopLatch.countDown();
+        }
+
+        public boolean awaitStart(long time, TimeUnit unit) throws InterruptedException
+        {
+            return startLatch.await(time, unit);
+        }
+
+
+        public boolean awaitStop(long time, TimeUnit unit) throws InterruptedException
+        {
+            return stopLatch.await(time, unit);
+        }
+    }
 }