Corrections to SingleThreadedEventHandlingStrategy, SocketAcceptor/Initiator (#201)

chrjohn · web-flow · commit 21bec091ded4 · 2018-06-21T09:39:28.000+02:00
* Corrections to SingleThreadedEventHandlingStrategy and SocketAcceptor/Initiator. ** Reverted changes from #192 since we actually want to process all messages on stop(). * Return from block() when stopped. ** It does not make sense to call getMessage() in that case. * minor cleanup in ThreadPerSessionEventHandlingStrategy * Added javadoc and unit test... ** to ensure that the session timer is not stopped by Message Processor thread when using stopHandlingMessages(true) * Changed SocketAcceptor and SocketInitiator to use SingleThreadedEventHandlingStrategy.stopHandlingMessages(true). ** Otherwise it could happen on quick restarts that the Message Processor thread stopped the SessionTimer concurrently which lead to an unresponsive FIX session, i.e. no Logon sent, no Heartbeats.
diff --git a/quickfixj-core/src/main/java/quickfix/SocketAcceptor.java b/quickfixj-core/src/main/java/quickfix/SocketAcceptor.java
@@ -122,7 +122,7 @@ public void stop(boolean forceDisconnect) {
                 stopSessionTimer();
             } finally {
                 try {
-                    eventHandlingStrategy.stopHandlingMessages();
+                    eventHandlingStrategy.stopHandlingMessages(true);
                 } finally {
                     Session.unregisterSessions(getSessions(), true);
                     clearConnectorSessions();
diff --git a/quickfixj-core/src/main/java/quickfix/SocketInitiator.java b/quickfixj-core/src/main/java/quickfix/SocketInitiator.java
@@ -134,7 +134,7 @@ public void stop(boolean forceDisconnect) {
                 stopInitiators();
             } finally {
                 try {
-                    eventHandlingStrategy.stopHandlingMessages();
+                    eventHandlingStrategy.stopHandlingMessages(true);
                 } finally {
                     Session.unregisterSessions(getSessions(), true);
                     clearConnectorSessions();
diff --git a/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java b/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java
@@ -328,6 +328,14 @@ protected void stopSessionTimer() {
         }
     }
 
+    // visible for testing
+    boolean checkSessionTimerRunning() {
+        if ( sessionTimerFuture != null ) {
+            return !sessionTimerFuture.isDone();
+        }
+        return false;
+    }
+
     protected ScheduledExecutorService getScheduledExecutorService() {
         return scheduledExecutorService;
     }
diff --git a/quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java b/quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java
@@ -99,8 +99,8 @@ private void block() {
                         sessionConnector.stopSessionTimer();
                         // reset the stoptime
                         stopTime = 0;
-                        return;
                     }
+                    return;
                 }
             }
             try {
@@ -167,16 +167,27 @@ private synchronized void startHandlingMessages() {
         isStopped = false;
     }
 
+    /**
+     * Stops processing of messages without waiting for message processing
+     * thread to finish.
+     * 
+     * It is advised to call stopHandlingMessages(true) instead of this method.
+     */
     public synchronized void stopHandlingMessages() {
         for (Session session : sessionConnector.getSessionMap().values()) {
             onMessage(session, END_OF_STREAM);
         }
         isStopped = true;
     }
 
+    /**
+     * Stops processing of messages and optionally waits for message processing
+     * thread to finish.
+     *
+     * @param join true to wait for thread to finish
+     */
     public void stopHandlingMessages(boolean join) {
         stopHandlingMessages();
-        messageProcessingThread.interrupt();
         if (join) {
             try {
                 messageProcessingThread.join();
@@ -227,12 +238,6 @@ public void start() {
                     executor.execute(wrapper);
 		}
                 
-                public void interrupt() {
-                    if (executor instanceof DedicatedThreadExecutor) {
-                        ((DedicatedThreadExecutor)executor).interrupt();
-                    }
-                }
-
 		/**
 		 * Provides the Thread::join and Thread::isAlive semantics on the nested Runnable.
 		 */
@@ -290,12 +295,6 @@ public void execute(Runnable command) {
 				thread.setDaemon(true);
 				thread.start();
 			}
-                        
-                        public void interrupt() {
-                            if (thread != null) {
-                                thread.interrupt();
-                            }
-                        }
 		}
 
 	}
diff --git a/quickfixj-core/src/main/java/quickfix/mina/ThreadPerSessionEventHandlingStrategy.java b/quickfixj-core/src/main/java/quickfix/mina/ThreadPerSessionEventHandlingStrategy.java
@@ -113,59 +113,58 @@ public void stopDispatcherThreads() {
         }
     }
 
-	/**
-	 * A stand-in for the Thread class that delegates to an Executor.
-	 * Implements all the API required by pre-existing QFJ code.
-	 */
-	protected static abstract class ThreadAdapter implements Runnable {
-
-		private final Executor executor;
-		private final String name;
-
-		public ThreadAdapter(String name, Executor executor) {
-			this.name = name;
-			this.executor = executor != null ? executor : new DedicatedThreadExecutor(name);
-		}
-
-		public void start() {
-			executor.execute(this);
-		}
-
-		@Override
-		public final void run() {
-			Thread currentThread = Thread.currentThread();
-			String threadName = currentThread.getName();
-			try {
-				if (!name.equals(threadName)) {
-					currentThread.setName(name + " (" + threadName + ")");
-				}
-				doRun();
-			} finally {
-				currentThread.setName(threadName);
-			}
-		}
-
-		abstract void doRun();
-
-		/**
-		 * An Executor that uses it's own dedicated Thread.
-		 * Provides equivalent behavior to the prior non-Executor approach.
-		 */
-		static final class DedicatedThreadExecutor implements Executor {
-
-			private final String name;
-			
-			DedicatedThreadExecutor(String name) {
-				this.name = name;
-			}
-
-			@Override
-			public void execute(Runnable command) {
-				new Thread(command, name).start();
-			}
-
-		}
+    /**
+     * A stand-in for the Thread class that delegates to an Executor.
+     * Implements all the API required by pre-existing QFJ code.
+     */
+    protected static abstract class ThreadAdapter implements Runnable {
+
+        private final Executor executor;
+        private final String name;
 
+        public ThreadAdapter(String name, Executor executor) {
+            this.name = name;
+            this.executor = executor != null ? executor : new DedicatedThreadExecutor(name);
+        }
+
+        public void start() {
+            executor.execute(this);
+        }
+
+        @Override
+        public final void run() {
+            Thread currentThread = Thread.currentThread();
+            String threadName = currentThread.getName();
+            try {
+                if (!name.equals(threadName)) {
+                    currentThread.setName(name + " (" + threadName + ")");
+                }
+                doRun();
+            } finally {
+                currentThread.setName(threadName);
+            }
+        }
+
+        abstract void doRun();
+
+        /**
+         * An Executor that uses its own dedicated Thread. Provides equivalent
+         * behavior to the prior non-Executor approach.
+         */
+        static final class DedicatedThreadExecutor implements Executor {
+
+            private final String name;
+
+            DedicatedThreadExecutor(String name) {
+                this.name = name;
+            }
+
+            @Override
+            public void execute(Runnable command) {
+                new Thread(command, name).start();
+            }
+
+        }
 	}
 
 	protected class MessageDispatchingThread extends ThreadAdapter {
@@ -226,7 +225,7 @@ void doRun() {
                 }
             }
             if (!messages.isEmpty()) {
-                final List<Message> tempList = new ArrayList<>();
+                final List<Message> tempList = new ArrayList<>(messages.size());
                 queueTracker.drainTo(tempList);
                 for (Message message : tempList) {
                     try {
diff --git a/quickfixj-core/src/test/java/quickfix/mina/SingleThreadedEventHandlingStrategyTest.java b/quickfixj-core/src/test/java/quickfix/mina/SingleThreadedEventHandlingStrategyTest.java
@@ -45,6 +45,7 @@
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import org.junit.AfterClass;
+import static org.junit.Assert.assertTrue;
 
 /**
  *
@@ -149,6 +150,23 @@ public void testMultipleStartStop() throws Exception {
         }
     }
 
+    /**
+     * During quick restarts: make sure that session timer is started and not stopped via
+     * block() method called from Message Processor thread.
+     */
+    @Test
+    public void testMultipleStartSessionTimer() throws Exception {
+        SessionSettings settings = new SessionSettings();
+        SessionConnector connector = new SessionConnectorUnderTest(settings, sessionFactory);
+        ehs = new SingleThreadedEventHandlingStrategy(connector, 1000);
+        for (int i = 0; i < 1000; i++) {
+            connector.startSessionTimer();
+            ehs.blockInThread();
+            assertTrue(connector.checkSessionTimerRunning());
+            ehs.stopHandlingMessages(true);
+        }
+    }
+
     @Test
     public void shouldCleanUpAcceptorQFJMessageProcessorThreadAfterInterrupt() throws Exception {
         assertQFJMessageProcessorThreads(0);

Original file line number	Diff line number	Diff line change
`@@ -328,6 +328,14 @@ protected void stopSessionTimer() {`
`328`	`328`	`}`
`329`	`329`	`}`
`330`	`330`
	`331`	`+ // visible for testing`
	`332`	`+ boolean checkSessionTimerRunning() {`
	`333`	`+ if ( sessionTimerFuture != null ) {`
	`334`	`+ return !sessionTimerFuture.isDone();`
	`335`	`+ }`
	`336`	`+ return false;`
	`337`	`+ }`
	`338`	`+`
`331`	`339`	`protected ScheduledExecutorService getScheduledExecutorService() {`
`332`	`340`	`return scheduledExecutorService;`
`333`	`341`	`}`