analyze flaky TimerTest

.github/workflows/maven.yml

@@ -17,12 +17,21 @@ jobs:
     name: Test JDK ${{ matrix.java }}, ${{ matrix.os }}
 
     steps:
-      - uses: actions/checkout@v1
+      - uses: actions/checkout@v2
+
       - name: Set up JDK
-        uses: actions/setup-java@v1
+        uses: actions/setup-java@v2
         with:
+          distribution: 'adopt'
           java-version: ${{ matrix.java }}
+
+      - name: Disable NTP on macOS (https://github.com/actions/virtual-environments/issues/820)
+        run: |
+          sudo systemsetup -setusingnetworktime off
+          sudo rm -rf /etc/ntp.conf
+        if: runner.os == 'macOS'
+
       - name: Test with Maven
         env:
           MAVEN_OPTS: "-Xms3g -Xmx3g"
-        run: mvn test -B -V -D"java.util.logging.config.file"="logging.properties" -D"http.keepAlive"="false" -D"maven.wagon.http.pool"="false" -D"maven.wagon.httpconnectionManager.ttlSeconds"="120"
+        run: mvn test -B -V -D"http.keepAlive"="false" -D"maven.wagon.http.pool"="false" -D"maven.wagon.httpconnectionManager.ttlSeconds"="120"

README.md

@@ -6,6 +6,7 @@ QuickFIX/J
 [![Total alerts](https://img.shields.io/lgtm/alerts/g/quickfix-j/quickfixj.svg?logo=lgtm&logoWidth=18)](https://lgtm.com/projects/g/quickfix-j/quickfixj/alerts/)
 [![Language grade: Java](https://img.shields.io/lgtm/grade/java/g/quickfix-j/quickfixj.svg?logo=lgtm&logoWidth=18)](https://lgtm.com/projects/g/quickfix-j/quickfixj/context:java)
 
+
 This is the official QuickFIX/J project repository.
 
 ## intro

quickfixj-core/src/main/java/quickfix/Session.java

@@ -760,7 +760,7 @@ private void setEnabled(boolean enabled) {
     }
 
     private void initializeHeader(Message.Header header) {
-        state.setLastSentTime(SystemTime.currentTimeMillis());
+//        state.setLastSentTime(SystemTime.currentTimeMillis());  move to sendRaw()
         header.setString(BeginString.FIELD, sessionID.getBeginString());
         header.setString(SenderCompID.FIELD, sessionID.getSenderCompID());
         optionallySetID(header, SenderSubID.FIELD, sessionID.getSenderSubID());
@@ -1753,6 +1753,7 @@ private boolean verify(Message msg, boolean checkTooHigh, boolean checkTooLow)
             UnsupportedMessageType, IOException {
 
         state.setLastReceivedTime(SystemTime.currentTimeMillis());
+        state.setLastReceivedTimeNanos(SystemTime.currentTimeMillisFromNanos());
         state.clearTestRequestCounter();
 
         String msgType;
@@ -1975,11 +1976,11 @@ public void next() throws IOException {
                 LOG.warn("Heartbeat failure detected but deactivated");
             }
         } else {
-            if (state.isTestRequestNeeded()) {
+            if (state.isTestRequestNeeded(sessionID)) {
                 generateTestRequest("TEST");
                 getLog().onEvent("Sent test request TEST");
                 stateListener.onMissedHeartBeat();
-            } else if (state.isHeartBeatNeeded()) {
+            } else if (state.isHeartBeatNeeded(sessionID)) {
                 generateHeartbeat();
             }
         }
@@ -2036,6 +2037,7 @@ private boolean generateLogon() throws IOException {
             logon.setBoolean(ResetSeqNumFlag.FIELD, true);
         }
         state.setLastReceivedTime(SystemTime.currentTimeMillis());
+        state.setLastReceivedTimeNanos(SystemTime.currentTimeMillisFromNanos());
         state.clearTestRequestCounter();
         state.setLogonSent(true);
         logonAttempts++;
@@ -2336,6 +2338,7 @@ private void resendMessages(Message receivedMessage, int beginSeqNo, int endSeqN
                         generateSequenceReset(receivedMessage, begin, msgSeqNum);
                     }
                     getLog().onEvent("Resending message: " + msgSeqNum);
+                    state.setLastSentTime(SystemTime.currentTimeMillis());
                     send(msg.toString());
                     begin = 0;
                     appMessageJustSent = true;
@@ -2591,8 +2594,8 @@ private boolean sendRaw(Message message, int num) {
             final Message.Header header = message.getHeader();
             final String msgType = header.getString(MsgType.FIELD);
 
-            initializeHeader(header);
-
+            initializeHeader(header);   // TODO still duplicate to generateXXX methods
+            state.setLastSentTime(SystemTime.currentTimeMillis());
             if (num > 0) {
                 header.setInt(MsgSeqNum.FIELD, num);
             }
@@ -2647,7 +2650,6 @@ private boolean sendRaw(Message message, int num) {
                     result = send(messageString);
                 }
             }
-
             return result;
         } catch (final IOException e) {
             logThrowable(getLog(), "Error reading/writing in MessageStore", e);

quickfixj-core/src/main/java/quickfix/SessionState.java

@@ -54,7 +54,9 @@ public final class SessionState {
     private boolean logoutReceived = false;
     private int testRequestCounter;
     private long lastSentTime;
+    private long lastSentTimeNanos;
     private long lastReceivedTime;
+    private long lastReceivedTimeNanos;
     private final double testRequestDelayMultiplier;
     private final double heartBeatTimeoutMultiplier;
     private long heartBeatMillis = Long.MAX_VALUE;
@@ -112,10 +114,23 @@ long getHeartBeatMillis() {
         }
     }
 
-    public boolean isHeartBeatNeeded() {
-        long millisSinceLastSentTime = SystemTime.currentTimeMillis() - getLastSentTime();
+    SessionID sessionID1 = new SessionID(FixVersions.BEGINSTRING_FIX44, "ISLD", "TW");
+//    SessionID sessionID2 = new SessionID(FixVersions.BEGINSTRING_FIX44, "TW", "ISLD");
+    public boolean isHeartBeatNeeded(SessionID sessionID) {
+        long currentTimeMillisFromNanos = SystemTime.currentTimeMillisFromNanos();
+        long lastSentTimeNanos = getLastSentTimeNanos();
+        long millisSinceLastSentTime = TimeUnit.NANOSECONDS.toMillis(currentTimeMillisFromNanos - lastSentTimeNanos);
+//        long millisSinceLastSentTime = SystemTime.currentTimeMillis() - getLastSentTime();
+        boolean name = millisSinceLastSentTime + 10 > getHeartBeatMillis() && getTestRequestCounter() == 0;
         // QFJ-448: allow 10 ms leeway since exact comparison causes skipped heartbeats occasionally
-        return millisSinceLastSentTime + 10 > getHeartBeatMillis() && getTestRequestCounter() == 0;
+        if (sessionID1.getSenderCompID().equals(sessionID.getSenderCompID())
+                || sessionID1.getSenderCompID().equals(sessionID.getTargetCompID())
+                || "ACCEPTOR-1".equals(sessionID.getTargetCompID())) {
+            getLog().onEvent("isHeartBeatNeeded() = " + name + " current=" + currentTimeMillisFromNanos
+                    + " - lastSentTime=" + lastSentTimeNanos
+                    + "= millisSinceLastSent=" + millisSinceLastSentTime);
+        }
+        return name;
     }
 
     public boolean isInitiator() {
@@ -128,21 +143,40 @@ public long getLastReceivedTime() {
         }
     }
 
+    public long getLastReceivedTimeNanos() {
+        synchronized (lock) {
+            return lastReceivedTimeNanos;
+        }
+    }
+
     public void setLastReceivedTime(long lastReceivedTime) {
         synchronized (lock) {
             this.lastReceivedTime = lastReceivedTime;
         }
     }
 
+    public void setLastReceivedTimeNanos(long lastReceivedTimeNanos) {
+        synchronized (lock) {
+            this.lastReceivedTimeNanos = lastReceivedTimeNanos;
+        }
+    }
+
     public long getLastSentTime() {
         synchronized (lock) {
             return lastSentTime;
         }
     }
 
+    public long getLastSentTimeNanos() {
+        synchronized (lock) {
+            return lastSentTimeNanos;
+        }
+    }
+
     public void setLastSentTime(long lastSentTime) {
         synchronized (lock) {
             this.lastSentTime = lastSentTime;
+            this.lastSentTimeNanos = SystemTime.currentTimeMillisFromNanos();
         }
     }
 
@@ -184,7 +218,8 @@ public void setLogonSent(boolean logonSent) {
 
     public boolean isLogonTimedOut() {
         synchronized (lock) {
-            return isLogonSent() && SystemTime.currentTimeMillis() - getLastReceivedTime() >= getLogonTimeoutMs();
+//            return isLogonSent() && SystemTime.currentTimeMillis() - getLastReceivedTime() >= getLogonTimeoutMs();
+            return isLogonSent() && TimeUnit.NANOSECONDS.toMillis(SystemTime.currentTimeMillisFromNanos() - getLastReceivedTimeNanos()) >= getLogonTimeoutMs();
         }
     }
 
@@ -282,14 +317,22 @@ public void incrementTestRequestCounter() {
         }
     }
 
-    public boolean isTestRequestNeeded() {
+    public boolean isTestRequestNeeded(SessionID sessionID) {
         long millisSinceLastReceivedTime = timeSinceLastReceivedMessage();
+        if (sessionID.getSenderCompID().equals(sessionID1.getSenderCompID())
+                || sessionID.getTargetCompID().equals(sessionID1.getSenderCompID())
+                || "ACCEPTOR-1".equals(sessionID.getTargetCompID())) {
+            getLog().onEvent("isTestRequestNeeded() - millisSinceLastReceived = " + millisSinceLastReceivedTime);
+        }
         return millisSinceLastReceivedTime >= ((1 + testRequestDelayMultiplier) * (getTestRequestCounter() + 1))
                 * getHeartBeatMillis();
     }
 
     private long timeSinceLastReceivedMessage() {
-        return SystemTime.currentTimeMillis() - getLastReceivedTime();
+        long lastReceivedTimeNanos1 = getLastReceivedTimeNanos();
+        long currentTimeMillisFromNanos = SystemTime.currentTimeMillisFromNanos();
+        long timeSinceLastReceivedMessage = TimeUnit.NANOSECONDS.toMillis(currentTimeMillisFromNanos - lastReceivedTimeNanos1);
+        return timeSinceLastReceivedMessage;
     }
 
     public boolean isTimedOut() {

quickfixj-core/src/main/java/quickfix/SystemTime.java

@@ -37,6 +37,11 @@ public long getTime() {
             return System.currentTimeMillis();
         }
 
+        @Override
+        public long getTimeFromNanos() {
+            return System.nanoTime();
+        }
+
         @Override
         public LocalDateTime getNow() {
             return LocalDateTime.now(ZoneOffset.UTC);
@@ -48,6 +53,10 @@ public LocalDateTime getNow() {
     public static long currentTimeMillis() {
         return systemTimeSource.getTime();
     }
+
+    public static long currentTimeMillisFromNanos() {
+        return systemTimeSource.getTimeFromNanos();
+    }
 
     public static LocalDateTime now() {
         return systemTimeSource.getNow();

quickfixj-core/src/main/java/quickfix/SystemTimeSource.java

@@ -33,6 +33,8 @@ public interface SystemTimeSource {
      * @return current (possible simulated) time
      */
     long getTime();
+
+    long getTimeFromNanos();
 
     /**
      * Obtain current LocalDateTime.

quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java

@@ -53,6 +53,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import org.apache.mina.core.future.CloseFuture;
 import org.apache.mina.core.service.IoService;
+import quickfix.SystemTime;
 
 /**
  * An abstract base class for acceptors and initiators. Provides support for common functionality and also serves as an
@@ -265,15 +266,15 @@ protected void logoutAllSessions(boolean forceDisconnect) {
     }
 
     protected void waitForLogout() {
-        long start = System.currentTimeMillis();
+        long start = SystemTime.currentTimeMillisFromNanos();
         Set<Session> loggedOnSessions;
         while (!(loggedOnSessions = getLoggedOnSessions()).isEmpty()) {
             try {
                 Thread.sleep(100L);
             } catch (InterruptedException e) {
                 log.error(e.getMessage(), e);
             }
-            final long elapsed = System.currentTimeMillis() - start;
+            final long elapsed = TimeUnit.NANOSECONDS.toMillis(SystemTime.currentTimeMillisFromNanos() - start);
             Iterator<Session> sessionItr = loggedOnSessions.iterator();
             while (sessionItr.hasNext()) {
                 Session session = sessionItr.next();

quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java

@@ -76,7 +76,7 @@ public void onMessage(Session quickfixSession, Message message) {
             queueTracker.put(new SessionMessageEvent(quickfixSession, message));
         } catch (InterruptedException e) {
             isStopped = true;
-            throw new RuntimeException(e);
+            Thread.currentThread().interrupt();
         }
     }
 
@@ -194,7 +194,9 @@ public void stopHandlingMessages(boolean join) {
         stopHandlingMessages();
         if (join) {
             try {
-                messageProcessingThread.join();
+                if (messageProcessingThread != null) {
+                    messageProcessingThread.join();
+                }
             } catch (InterruptedException e) {
                 sessionConnector.log.error("{} interrupted.", MESSAGE_PROCESSOR_THREAD_NAME);
             }

quickfixj-core/src/test/java/quickfix/MockSystemTimeSource.java

@@ -23,9 +23,12 @@
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.util.Calendar;
+import java.util.concurrent.TimeUnit;
 
 public class MockSystemTimeSource implements SystemTimeSource {
-    private long[] systemTimes = { System.currentTimeMillis() };
+
+    private long[] systemTimes = {System.currentTimeMillis()};
+    private long[] systemTimesNanos = {System.nanoTime()};
     private int offset;
 
     public MockSystemTimeSource() {
@@ -34,14 +37,19 @@ public MockSystemTimeSource() {
 
     public MockSystemTimeSource(long time) {
         setSystemTimes(time);
+        setSystemTimesNanos(TimeUnit.MILLISECONDS.toNanos(time));
     }
 
     public void setSystemTimes(long[] times) {
         systemTimes = times;
     }
 
     void setSystemTimes(long time) {
-        systemTimes = new long[] { time };
+        systemTimes = new long[]{time};
+    }
+
+    void setSystemTimesNanos(long time) {
+        systemTimesNanos = new long[]{time};
     }
 
     public void setTime(Calendar c) {
@@ -56,10 +64,21 @@ public long getTime() {
         return systemTimes[offset];
     }
 
+    @Override
+    public long getTimeFromNanos() {
+        if (systemTimesNanos.length - offset > 1) {
+            offset++;
+        }
+        return systemTimesNanos[offset];
+    }
+
     public void increment(long delta) {
         if (systemTimes.length - offset == 1) {
             systemTimes[offset] += delta;
         }
+        if (systemTimesNanos.length - offset == 1) {
+            systemTimesNanos[offset] += TimeUnit.MILLISECONDS.toNanos(delta);
+        }
     }
 
     @Override