Skip to content

QFJ-375 - configurable timeout multiplier #288

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
May 19, 2020
Merged

QFJ-375 - configurable timeout multiplier #288

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
810ba9d
QFJ-375 - configurable timeout multiplier
the-thing May 14, 2020
de4a975
QFJ-375 - configurable timeout multiplier
the-thing May 19, 2020
42aea7b
QFJ-375 - replaced heartbeat timeout multiplier magic number with def…
the-thing May 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions quickfixj-core/src/main/doc/usermanual/usage/configuration.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1285,11 +1285,21 @@ <H3>QuickFIX Settings</H3>
<TR ALIGN="left" VALIGN="middle">
<TD><I>TestRequestDelayMultiplier</I></TD>
<TD>Fraction of the heartbeat interval which defines the additional time to wait
if a TestRequest sent after a missing heartbeat times out.
if a TestRequest sent after a missing heartbeat times out (final coefficient value is equal to
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know if I made it worst by clarifying the final coefficient values as for TestRequestDelayMultiplier there is also number of requests variable which might require to look in in the code anyway.

TestRequestDelayMultiplier + 1.0).
</TD>
<TD>0..1</TD>
<TD>any non-negative value</TD>
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This has only informative value as there were no checks anyway.

<TD>0.5</TD>
</TR>
<TR ALIGN="left" VALIGN="middle">
<TD><I>HeartBeatTimeoutMultiplier</I></TD>
<TD>Fraction of the heartbeat interval which defines the additional time to wait
since the last message was received before disconnecting (final coefficient value is equal to
HeartBeatTimeoutMultiplier + 1.0).
</TD>
<TD>any non-negative value</TD>
<TD>1.4</TD>
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The default hardcoded values was 2.4, but +1 is always done in the code the same as for TestRequestDelayMultiplier

</TR>
<TR ALIGN="left" VALIGN="middle">
<TD><I>DisableHeartBeatCheck</I></TD>
<TD> Heartbeat detection is disabled. A disconnect due to a missing heartbeat will never occur.</TD>
Expand Down
5 changes: 4 additions & 1 deletion quickfixj-core/src/main/java/quickfix/DefaultSessionFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,9 @@ public Session create(SessionID sessionID, SessionSettings settings) throws Conf
final double testRequestDelayMultiplier = getSetting(settings, sessionID,
Session.SETTING_TEST_REQUEST_DELAY_MULTIPLIER,
Session.DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER);
final double heartBeatTimeoutMultiplier = getSetting(settings, sessionID,
Session.SETTING_HEARTBEAT_TIMEOUT_MULTIPLIER,
Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);

final UtcTimestampPrecision timestampPrecision = getTimestampPrecision(settings, sessionID,
UtcTimestampPrecision.MILLIS);
Expand Down Expand Up @@ -228,7 +231,7 @@ public Session create(SessionID sessionID, SessionSettings settings) throws Conf
rejectInvalidMessage, rejectMessageOnUnhandledException, requiresOrigSendingTime,
forceResendWhenCorruptedStore, allowedRemoteAddresses, validateIncomingMessage,
resendRequestChunkSize, enableNextExpectedMsgSeqNum, enableLastMsgSeqNumProcessed,
validateChecksum, logonTags);
validateChecksum, logonTags, heartBeatTimeoutMultiplier);

session.setLogonTimeout(logonTimeout);
session.setLogoutTimeout(logoutTimeout);
Expand Down
31 changes: 18 additions & 13 deletions quickfixj-core/src/main/java/quickfix/Session.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,10 +113,15 @@ public class Session implements Closeable {
public static final String SETTING_MAX_LATENCY = "MaxLatency";

/**
* Session setting for the test delay multiplier (0-1, as fraction of Heartbeat interval)
* Session setting for the test delay multiplier (as fraction of heartbeat interval).
*/
public static final String SETTING_TEST_REQUEST_DELAY_MULTIPLIER = "TestRequestDelayMultiplier";

/**
* Session setting for the heartbeat timeout multiplier (as fraction of heartbeat interval).
*/
public static final String SETTING_HEARTBEAT_TIMEOUT_MULTIPLIER = "HeartBeatTimeoutMultiplier";

/**
* Session scheduling setting to specify that session never reset
*/
Expand Down Expand Up @@ -426,14 +431,15 @@ public class Session implements Closeable {

private final AtomicReference<ApplVerID> targetDefaultApplVerID = new AtomicReference<>();
private final DefaultApplVerID senderDefaultApplVerID;
private boolean validateSequenceNumbers = true;
private boolean validateIncomingMessage = true;
private final boolean validateSequenceNumbers;
private final boolean validateIncomingMessage;
private final int[] logonIntervals;
private final Set<InetAddress> allowedRemoteAddresses;

public static final int DEFAULT_MAX_LATENCY = 120;
public static final int DEFAULT_RESEND_RANGE_CHUNK_SIZE = 0; // no resend range
public static final double DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER = 0.5;
public static final double DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER = 1.4;
private static final String ENCOUNTERED_END_OF_STREAM = "Encountered END_OF_STREAM";


Expand All @@ -447,15 +453,14 @@ public class Session implements Closeable {

protected static final Logger LOG = LoggerFactory.getLogger(Session.class);


Session(Application application, MessageStoreFactory messageStoreFactory, SessionID sessionID,
DataDictionaryProvider dataDictionaryProvider, SessionSchedule sessionSchedule,
LogFactory logFactory, MessageFactory messageFactory, int heartbeatInterval) {
this(application, messageStoreFactory, sessionID, dataDictionaryProvider, sessionSchedule,
logFactory, messageFactory, heartbeatInterval, true, DEFAULT_MAX_LATENCY, UtcTimestampPrecision.MILLIS,
false, false, false, false, true, false, true, false,
DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER, null, true, new int[]{5}, false, false,
false, false, true, false, true, false, null, true, DEFAULT_RESEND_RANGE_CHUNK_SIZE, false, false, false, new ArrayList<StringField>());
DataDictionaryProvider dataDictionaryProvider, SessionSchedule sessionSchedule, LogFactory logFactory,
MessageFactory messageFactory, int heartbeatInterval) {
this(application, messageStoreFactory, sessionID, dataDictionaryProvider, sessionSchedule, logFactory,
messageFactory, heartbeatInterval, true, DEFAULT_MAX_LATENCY, UtcTimestampPrecision.MILLIS, false, false,
false, false, true, false, true, false, DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER, null, true, new int[] {5},
false, false, false, false, true, false, true, false, null, true, DEFAULT_RESEND_RANGE_CHUNK_SIZE, false,
false, false, new ArrayList<StringField>(), DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);
}

Session(Application application, MessageStoreFactory messageStoreFactory, SessionID sessionID,
Expand All @@ -473,7 +478,7 @@ public class Session implements Closeable {
boolean forceResendWhenCorruptedStore, Set<InetAddress> allowedRemoteAddresses,
boolean validateIncomingMessage, int resendRequestChunkSize,
boolean enableNextExpectedMsgSeqNum, boolean enableLastMsgSeqNumProcessed,
boolean validateChecksum, List<StringField> logonTags) {
boolean validateChecksum, List<StringField> logonTags, double heartBeatTimeoutMultiplier) {
this.application = application;
this.sessionID = sessionID;
this.sessionSchedule = sessionSchedule;
Expand Down Expand Up @@ -520,7 +525,7 @@ public class Session implements Closeable {
}

state = new SessionState(this, engineLog, heartbeatInterval, heartbeatInterval != 0,
messageStore, testRequestDelayMultiplier);
messageStore, testRequestDelayMultiplier, heartBeatTimeoutMultiplier);

registerSession(this);

Expand Down
15 changes: 6 additions & 9 deletions quickfixj-core/src/main/java/quickfix/SessionState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
Expand Down Expand Up @@ -55,6 +56,7 @@ public final class SessionState {
private long lastSentTime;
private long lastReceivedTime;
private final double testRequestDelayMultiplier;
private final double heartBeatTimeoutMultiplier;
private long heartBeatMillis = Long.MAX_VALUE;
private int heartBeatInterval;

Expand All @@ -75,13 +77,14 @@ public final class SessionState {
private final Map<Integer, Message> messageQueue = new LinkedHashMap<>();

public SessionState(Object lock, Log log, int heartBeatInterval, boolean initiator, MessageStore messageStore,
double testRequestDelayMultiplier) {
double testRequestDelayMultiplier, double heartBeatTimeoutMultiplier) {
this.lock = lock;
this.initiator = initiator;
this.messageStore = messageStore;
setHeartBeatInterval(heartBeatInterval);
this.log = log == null ? new NullLog() : log;
this.testRequestDelayMultiplier = testRequestDelayMultiplier;
this.heartBeatTimeoutMultiplier = heartBeatTimeoutMultiplier;
}

public int getHeartBeatInterval() {
Expand All @@ -93,13 +96,7 @@ public int getHeartBeatInterval() {
public void setHeartBeatInterval(int heartBeatInterval) {
synchronized (lock) {
this.heartBeatInterval = heartBeatInterval;
}
setHeartBeatMillis(heartBeatInterval * 1000L);
}

private void setHeartBeatMillis(long heartBeatMillis) {
synchronized (lock) {
this.heartBeatMillis = heartBeatMillis;
this.heartBeatMillis = TimeUnit.SECONDS.toMillis(heartBeatInterval);
}
}

Expand Down Expand Up @@ -291,7 +288,7 @@ private long timeSinceLastReceivedMessage() {

public boolean isTimedOut() {
long millisSinceLastReceivedTime = timeSinceLastReceivedMessage();
return millisSinceLastReceivedTime >= 2.4 * getHeartBeatMillis();
return millisSinceLastReceivedTime >= (1 + heartBeatTimeoutMultiplier) * getHeartBeatMillis();
}

public boolean set(int sequence, String message) throws IOException {
Expand Down
2 changes: 1 addition & 1 deletion quickfixj-core/src/test/java/quickfix/MockSystemTimeSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public void setSystemTimes(long[] times) {
systemTimes = times;
}

private void setSystemTimes(long time) {
void setSystemTimes(long time) {
systemTimes = new long[] { time };
}

Expand Down
3 changes: 2 additions & 1 deletion quickfixj-core/src/test/java/quickfix/SessionFactoryTestSupport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@ public static final class Builder {
private boolean persistMessages = false;
private final boolean useClosedRangeForResend = false;
private final double testRequestDelayMultiplier = 1.5;
private final double heartBeatTimeoutMultiplier = 2.5;
private DefaultApplVerID senderDefaultApplVerID = null;
private boolean validateSequenceNumbers = true;
private final int[] logonIntervals = new int[]{5};
Expand Down Expand Up @@ -122,7 +123,7 @@ public Session build() {
resetOnError, disconnectOnError, disableHeartBeatCheck, false, rejectInvalidMessage,
rejectMessageOnUnhandledException, requiresOrigSendingTime, forceResendWhenCorruptedStore,
allowedRemoteAddresses, validateIncomingMessage, resendRequestChunkSize, enableNextExpectedMsgSeqNum,
enableLastMsgSeqNumProcessed, validateChecksum, logonTags);
enableLastMsgSeqNumProcessed, validateChecksum, logonTags, heartBeatTimeoutMultiplier);
}

public Builder setBeginString(final String beginString) {
Expand Down
54 changes: 42 additions & 12 deletions quickfixj-core/src/test/java/quickfix/SessionStateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,22 +19,32 @@

package quickfix;

import junit.framework.TestCase;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class SessionStateTest extends TestCase {
protected void setUp() throws Exception {
super.setUp();
MockSystemTimeSource mockTimeSource = new MockSystemTimeSource(1000);
SystemTime.setTimeSource(mockTimeSource);
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

public class SessionStateTest {

private MockSystemTimeSource timeSource;

@Before
public void setUp() {
timeSource = new MockSystemTimeSource(1000);
SystemTime.setTimeSource(timeSource);
}

protected void tearDown() throws Exception {
@After
public void tearDown() {
SystemTime.setTimeSource(null);
super.tearDown();
}

public void testTimeoutDefaultsAreNonzero() throws Exception {
SessionState state = new SessionState(new Object(), null, 0, false, null, Session.DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER);
@Test
public void testTimeoutDefaultsAreNonzero() {
SessionState state = new SessionState(new Object(), null, 0, false, null,
Session.DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER, Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);
state.setLastReceivedTime(900);
assertFalse("logon timeout not init'ed", state.isLogonTimedOut());

Expand All @@ -43,8 +53,10 @@ public void testTimeoutDefaultsAreNonzero() throws Exception {
assertFalse("logout timeout not init'ed", state.isLogoutTimedOut());
}

public void testTestRequestTiming() throws Exception {
SessionState state = new SessionState(new Object(), null, 0, false, null, Session.DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER);
@Test
public void testTestRequestTiming() {
SessionState state = new SessionState(new Object(), null, 0, false, null,
Session.DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER, Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);
state.setLastReceivedTime(950);
state.setHeartBeatInterval(50);
assertFalse("testRequest shouldn't be needed yet", state.isTestRequestNeeded());
Expand All @@ -57,4 +69,22 @@ public void testTestRequestTiming() throws Exception {
state.setHeartBeatInterval(3);
assertFalse("testRequest shouldn't be needed yet", state.isTestRequestNeeded());
}

@Test
public void testSessionTimeout() {
SessionState state = new SessionState(new Object(), null, 30, false, null,
Session.DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER, Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);

// session should timeout after 2.4 * 30 = 72 seconds
state.setLastReceivedTime(950_000);

timeSource.setSystemTimes(1_000_000L);
assertFalse("session is still valid", state.isTimedOut());

timeSource.setSystemTimes(1_021_999L);
assertFalse("session is still valid", state.isTimedOut());

timeSource.setSystemTimes(1_022_000L);
assertTrue("session timed out", state.isTimedOut());
}
}
18 changes: 9 additions & 9 deletions quickfixj-core/src/test/java/quickfix/SessionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ public void testDisposalOfFileResources() throws Exception {
new DefaultMessageFactory(), 30, false, 30, UtcTimestampPrecision.MILLIS, true, false,
false, false, false, false, true, false, 1.5, null, true,
new int[] { 5 }, false, false, false, false, true, false, true, false,
null, true, 0, false, false, true, new ArrayList<>())) {
null, true, 0, false, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER)) {
// Simulate socket disconnect
session.setResponder(null);
}
Expand Down Expand Up @@ -142,7 +142,7 @@ public void testNondisposableFileResources() throws Exception {
new DefaultMessageFactory(), 30, false, 30, UtcTimestampPrecision.MILLIS, true, false,
false, false, false, false, true, false, 1.5, null, true,
new int[] { 5 }, false, false, false, false, true, false, true, false,
null, true, 0, false, false, true, new ArrayList<>())) {
null, true, 0, false, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER)) {
// Simulate socket disconnect
session.setResponder(null);

Expand Down Expand Up @@ -2074,7 +2074,7 @@ private void testSequenceResetGapFillWithChunkSize(int chunkSize)
UtcTimestampPrecision.MILLIS, resetOnLogon, false, false, false, false, false, true,
false, 1.5, null, validateSequenceNumbers, new int[] { 5 },
false, false, false, false, true, false, true, false, null, true,
chunkSize, false, false, true, new ArrayList<>())) {
chunkSize, false, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER)) {

UnitTestResponder responder = new UnitTestResponder();
session.setResponder(responder);
Expand Down Expand Up @@ -2136,7 +2136,7 @@ public void correct_sequence_number_for_last_gap_fill_if_next_sender_sequence_nu
new DefaultMessageFactory(), 30, false, 30, UtcTimestampPrecision.MILLIS, resetOnLogon,
false, false, false, false, false, true, false, 1.5, null, validateSequenceNumbers,
new int[]{5}, false, false, false, false, true, false, true, false, null, true, 0,
false, false, true, new ArrayList<>());
false, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);

Responder mockResponder = mock(Responder.class);
when(mockResponder.send(anyString())).thenReturn(true);
Expand Down Expand Up @@ -2184,7 +2184,7 @@ public void correct_sequence_number_for_last_gap_fill_if_next_sender_sequence_nu
new DefaultMessageFactory(), 30, false, 30, UtcTimestampPrecision.MILLIS, resetOnLogon,
false, false, false, false, false, true, false, 1.5, null, validateSequenceNumbers,
new int[]{5}, false, false, false, false, true, false, true, false, null, true, 0,
enableNextExpectedMsgSeqNum, false, true, new ArrayList<>());
enableNextExpectedMsgSeqNum, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);

Responder mockResponder = mock(Responder.class);
when(mockResponder.send(anyString())).thenReturn(true);
Expand Down Expand Up @@ -2233,7 +2233,7 @@ public void testMsgSeqNumTooHighWithDisconnectOnError() throws Exception {
UtcTimestampPrecision.MILLIS, resetOnLogon, false, false, false, false, false, true,
false, 1.5, null, validateSequenceNumbers, new int[] { 5 },
false, disconnectOnError, false, false, true, false, true, false,
null, true, 0, false, false, true, new ArrayList<>())) {
null, true, 0, false, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER)) {

UnitTestResponder responder = new UnitTestResponder();
session.setResponder(responder);
Expand Down Expand Up @@ -2269,7 +2269,7 @@ public void testTimestampPrecision() throws Exception {
UtcTimestampPrecision.NANOS, resetOnLogon, false, false, false, false, false, true,
false, 1.5, null, validateSequenceNumbers, new int[] { 5 },
false, disconnectOnError, false, false, true, false, true, false,
null, true, 0, false, false, true, new ArrayList<>())) {
null, true, 0, false, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER)) {

UnitTestResponder responder = new UnitTestResponder();
session.setResponder(responder);
Expand Down Expand Up @@ -2321,7 +2321,7 @@ private void testLargeQueue(int N) throws Exception {
new DefaultMessageFactory(), isInitiator ? 30 : 0, false, 30, UtcTimestampPrecision.MILLIS, resetOnLogon,
false, false, false, false, false, true, false, 1.5, null, validateSequenceNumbers,
new int[]{5}, false, false, false, false, true, false, true, false, null, true, 0,
false, false, true, new ArrayList<>());
false, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);

UnitTestResponder responder = new UnitTestResponder();
session.setResponder(responder);
Expand Down Expand Up @@ -2437,7 +2437,7 @@ public void fromAdmin(Message message, SessionID sessionId) throws FieldNotFound
new DefaultMessageFactory(), isInitiator ? 30 : 0, false, 30, UtcTimestampPrecision.MILLIS, resetOnLogon,
false, false, false, false, false, true, false, 1.5, null, validateSequenceNumbers,
new int[]{5}, false, false, false, false, true, false, true, false, null, true, 0,
enableNextExpectedMsgSeqNum, false, true, new ArrayList<>());
enableNextExpectedMsgSeqNum, false, true, new ArrayList<>(), Session.DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER);
UnitTestResponder responder = new UnitTestResponder();
session.setResponder(responder);

Expand Down