You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by al...@apache.org on 2020/07/16 15:05:15 UTC
[ignite] branch ignite-2.9 updated: IGNITE-13134 : checking if
failureDetectionTimeout enabled. - Fixes #7916.
This is an automated email from the ASF dual-hosted git repository.
alexpl pushed a commit to branch ignite-2.9
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/ignite-2.9 by this push:
new 06364dc IGNITE-13134 : checking if failureDetectionTimeout enabled. - Fixes #7916.
06364dc is described below
commit 06364dcace6625133a147ed510a366841f6f0284
Author: Vladimir Steshin <vl...@gmail.com>
AuthorDate: Thu Jul 16 14:16:13 2020 +0300
IGNITE-13134 : checking if failureDetectionTimeout enabled. - Fixes #7916.
Signed-off-by: Sergey Chugunov <se...@gmail.com>
(cherry picked from commit 1ca486eb4b5a425e4721f3608047511412aa22ee)
---
.../spi/IgniteSpiOperationTimeoutHelper.java | 83 +++++++++++-----------
.../spi/communication/tcp/TcpCommunicationSpi.java | 2 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 2 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 79 ++++++++++++++------
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 11 +--
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 2 +-
6 files changed, 108 insertions(+), 71 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiOperationTimeoutHelper.java b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiOperationTimeoutHelper.java
index 48161ba..20a5ebf 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiOperationTimeoutHelper.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiOperationTimeoutHelper.java
@@ -33,17 +33,14 @@ public class IgniteSpiOperationTimeoutHelper {
// https://issues.apache.org/jira/browse/IGNITE-11221
// We need to reuse new logic ExponentialBackoffTimeout logic in TcpDiscovery instead of this class.
- /** */
- private long lastOperStartNanos;
+ /** Flag whether to use timeout. */
+ private final boolean timeoutEnabled;
- /** */
- private long timeout;
+ /** Time in nanos which cannot be reached for current operation. */
+ private final long timeoutThreshold;
- /** */
- private final boolean failureDetectionTimeoutEnabled;
-
- /** */
- private final long failureDetectionTimeout;
+ /** Keeps {@code true} if last call to {@link #nextTimeoutChunk(long)} has timeouted. {@code False} otherwise. */
+ private boolean lastOperationTimeouted;
/**
* Constructor.
@@ -52,9 +49,7 @@ public class IgniteSpiOperationTimeoutHelper {
* @param srvOp {@code True} if communicates with server node.
*/
public IgniteSpiOperationTimeoutHelper(IgniteSpiAdapter adapter, boolean srvOp) {
- failureDetectionTimeoutEnabled = adapter.failureDetectionTimeoutEnabled();
- failureDetectionTimeout = srvOp ? adapter.failureDetectionTimeout() :
- adapter.clientFailureDetectionTimeout();
+ this(adapter, srvOp, -1, -1);
}
/**
@@ -62,15 +57,26 @@ public class IgniteSpiOperationTimeoutHelper {
*
* @param adapter SPI adapter.
* @param srvOp {@code True} if communicates with server node.
- * @param lastOperStartNanos Time of last related operation in nanos.
+ * @param lastRelatedOperationTime Time of last related operation in nanos. Ignored if negative, 0 or
+ * {@code adapter.failureDetectionTimeoutEnabled()} is false.
+ * @param absoluteThreshold Absolute time threshold (nanos) which must not be reached. Ignored if negative or 0.
*/
- public IgniteSpiOperationTimeoutHelper(IgniteSpiAdapter adapter, boolean srvOp, long lastOperStartNanos) {
- this(adapter, srvOp);
+ public IgniteSpiOperationTimeoutHelper(IgniteSpiAdapter adapter, boolean srvOp, long lastRelatedOperationTime,
+ long absoluteThreshold) {
+ timeoutEnabled = adapter.failureDetectionTimeoutEnabled();
+
+ if (timeoutEnabled) {
+ long timeout = (lastRelatedOperationTime > 0 ? lastRelatedOperationTime : System.nanoTime()) +
+ U.millisToNanos(srvOp ? adapter.failureDetectionTimeout() : adapter.clientFailureDetectionTimeout());
- this.lastOperStartNanos = lastOperStartNanos;
+ if (absoluteThreshold > 0 && timeout > absoluteThreshold)
+ timeout = absoluteThreshold;
- if (lastOperStartNanos > 0)
- timeout = failureDetectionTimeout;
+ timeoutThreshold = timeout;
+ } else {
+ // Save absolute threshold if it is set.
+ timeoutThreshold = absoluteThreshold > 0 ? absoluteThreshold : 0;
+ }
}
/**
@@ -85,42 +91,39 @@ public class IgniteSpiOperationTimeoutHelper {
* this {@code IgniteSpiOperationTimeoutController}.
*/
public long nextTimeoutChunk(long dfltTimeout) throws IgniteSpiOperationTimeoutException {
- if (!failureDetectionTimeoutEnabled)
- return dfltTimeout;
+ long now = System.nanoTime();
- if (lastOperStartNanos == 0) {
- timeout = failureDetectionTimeout;
- lastOperStartNanos = System.nanoTime();
- }
+ long left;
+
+ if (timeoutEnabled)
+ left = timeoutThreshold - now;
else {
- long curNanos = System.nanoTime();
+ left = U.millisToNanos(dfltTimeout);
- timeout -= U.nanosToMillis(curNanos - lastOperStartNanos);
+ if (timeoutThreshold > 0 && now + left >= timeoutThreshold)
+ left = timeoutThreshold - now;
+ }
- lastOperStartNanos = curNanos;
+ if (left <= 0) {
+ lastOperationTimeouted = true;
- if (timeout <= 0)
- throw new IgniteSpiOperationTimeoutException("Network operation timed out. Increase " +
- "'failureDetectionTimeout' configuration property [failureDetectionTimeout="
- + failureDetectionTimeout + ']');
+ throw new IgniteSpiOperationTimeoutException("Network operation timed out.");
}
-
- return timeout;
+
+ return U.nanosToMillis(left);
}
/**
- * Checks whether the given {@link Exception} is generated because failure detection timeout has been reached.
+ * Checks whether the given {@link Exception} is a timeout-exception or the has been reached in last call to
+ * {@code nextTimeoutChunk(long)}.
*
- * @param e Exception.
- * @return {@code true} if failure detection timeout is reached, {@code false} otherwise.
+ * @param e Exception to check if is a timeout.
+ * @return {@code True} if the excaption is a timeout or failure timeout was reached. {@code False} otherwise.
*/
public boolean checkFailureTimeoutReached(Exception e) {
- if (!failureDetectionTimeoutEnabled)
- return false;
-
if (X.hasCause(e, IgniteSpiOperationTimeoutException.class, SocketTimeoutException.class, SocketException.class))
return true;
- return (timeout - U.millisSinceNanos(lastOperStartNanos) <= 0);
+ return lastOperationTimeouted;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index 18139fe..80fbca1 100755
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -3272,7 +3272,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
getSpiContext().messageFormatter());
}
catch (IgniteCheckedException e) {
- if (timeoutHelper.checkFailureTimeoutReached(e))
+ if (connectAttempts == 1 && timeoutHelper.checkFailureTimeoutReached(e))
throw e;
// Reconnect for the second time, if connection is not established.
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 77bd521..59c4607 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@ -801,7 +801,7 @@ class ClientImpl extends TcpDiscoveryImpl {
break;
}
- if (timeoutHelper.checkFailureTimeoutReached(e))
+ if (spi.failureDetectionTimeoutEnabled() && timeoutHelper.checkFailureTimeoutReached(e))
break;
if (!spi.failureDetectionTimeoutEnabled() && ++reconCnt == spi.getReconnectCount())
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index e6e2123..37af9aa 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -903,7 +903,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (!openedSock && reconCnt == 2)
break;
- if (timeoutHelper.checkFailureTimeoutReached(e))
+ if (spi.failureDetectionTimeoutEnabled() && timeoutHelper.checkFailureTimeoutReached(e))
break;
else if (!spi.failureDetectionTimeoutEnabled() && reconCnt == spi.getReconnectCount())
break;
@@ -1532,7 +1532,7 @@ class ServerImpl extends TcpDiscoveryImpl {
break;
}
- if (timeoutHelper.checkFailureTimeoutReached(e))
+ if (spi.failureDetectionTimeoutEnabled() && timeoutHelper.checkFailureTimeoutReached(e))
break;
if (!spi.failureDetectionTimeoutEnabled() && ++reconCnt == spi.getReconnectCount())
@@ -3349,8 +3349,12 @@ class ServerImpl extends TcpDiscoveryImpl {
while (true) {
if (sock == null) {
- if (timeoutHelper == null)
- timeoutHelper = new IgniteSpiOperationTimeoutHelper(spi, true);
+ // We re-create the helper here because it could be created earlier with wrong timeout on
+ // message sending like IgniteConfiguration.failureDetectionTimeout. Here we are in the
+ // state of conenction recovering and have to work with
+ // TcpDiscoverSpi.getEffectiveConnectionRecoveryTimeout()
+ if (timeoutHelper == null || sndState != null)
+ timeoutHelper = serverOperationTimeoutHelper(sndState, -1);
boolean success = false;
@@ -3389,6 +3393,8 @@ class ServerImpl extends TcpDiscoveryImpl {
// We should take previousNodeAlive flag into account only if we received the response from the correct node.
if (res.creatorNodeId().equals(next.id()) && res.previousNodeAlive() && sndState != null) {
+ sndState.checkTimeout();
+
// Remote node checked connection to it's previous and got success.
boolean previousNode = sndState.markLastFailedNodeAlive();
@@ -3498,13 +3504,20 @@ class ServerImpl extends TcpDiscoveryImpl {
onException("Failed to connect to next node [msg=" + msg + ", err=" + e + ']', e);
+ // Fastens failure detection.
+ if (sndState != null && sndState.checkTimeout()) {
+ segmentLocalNodeOnSendFail(failedNodes);
+
+ return; // Nothing to do here.
+ }
+
if (!openSock)
break; // Don't retry if we can not establish connection.
if (!spi.failureDetectionTimeoutEnabled() && ++reconCnt == spi.getReconnectCount())
break;
- if (timeoutHelper.checkFailureTimeoutReached(e))
+ if (spi.failureDetectionTimeoutEnabled() && timeoutHelper.checkFailureTimeoutReached(e))
break;
else if (!spi.failureDetectionTimeoutEnabled() && (e instanceof
SocketTimeoutException || X.hasCause(e, SocketTimeoutException.class))) {
@@ -3563,10 +3576,8 @@ class ServerImpl extends TcpDiscoveryImpl {
addFailedNodes(pendingMsg, failedNodes);
- if (timeoutHelper == null) {
- timeoutHelper = new IgniteSpiOperationTimeoutHelper(spi, true,
- lastRingMsgSentTime);
- }
+ if (timeoutHelper == null)
+ timeoutHelper = serverOperationTimeoutHelper(sndState, lastRingMsgSentTime);
try {
spi.writeToSocket(sock, out, pendingMsg, timeoutHelper.nextTimeoutChunk(
@@ -3610,7 +3621,7 @@ class ServerImpl extends TcpDiscoveryImpl {
long tsNanos = System.nanoTime();
if (timeoutHelper == null)
- timeoutHelper = new IgniteSpiOperationTimeoutHelper(spi, true, lastRingMsgSentTime);
+ timeoutHelper = serverOperationTimeoutHelper(sndState, lastRingMsgSentTime);
addFailedNodes(msg, failedNodes);
@@ -3677,7 +3688,7 @@ class ServerImpl extends TcpDiscoveryImpl {
onException("Failed to send message to next node [next=" + next.id() + ", msg=" + msg + ']',
e);
- if (timeoutHelper.checkFailureTimeoutReached(e))
+ if (spi.failureDetectionTimeoutEnabled() && timeoutHelper.checkFailureTimeoutReached(e))
break;
if (!spi.failureDetectionTimeoutEnabled()) {
@@ -3714,6 +3725,11 @@ class ServerImpl extends TcpDiscoveryImpl {
if (!sent) {
if (sndState == null && spi.getEffectiveConnectionRecoveryTimeout() > 0)
sndState = new CrossRingMessageSendState();
+ else if (sndState != null && sndState.checkTimeout()) {
+ segmentLocalNodeOnSendFail(failedNodes);
+
+ return; // Nothing to do here.
+ }
boolean failedNextNode = sndState == null || sndState.markNextNodeFailed();
@@ -3747,12 +3763,6 @@ class ServerImpl extends TcpDiscoveryImpl {
}
}
- if (sndState != null && sndState.isFailed()) {
- segmentLocalNodeOnSendFail(failedNodes);
-
- return; // Nothing to do here.
- }
-
next = null;
errs = null;
@@ -6220,6 +6230,19 @@ class ServerImpl extends TcpDiscoveryImpl {
}
}
+ /**
+ * Creates proper timeout helper taking in account current send state.
+ *
+ * @param sndState Current connection recovering state. Ignored if {@code null}.
+ * @param lastOperationNanos Time of last related operation. Ignored if negative or 0.
+ * @return Timeout helper.
+ */
+ private IgniteSpiOperationTimeoutHelper serverOperationTimeoutHelper(@Nullable CrossRingMessageSendState sndState,
+ long lastOperationNanos) {
+ return new IgniteSpiOperationTimeoutHelper(spi, true, lastOperationNanos,
+ sndState == null ? -1 : sndState.failTimeNanos);
+ }
+
/** Fixates time of last sent message. */
private void updateLastSentMessageTime() {
lastRingMsgSentTime = System.nanoTime();
@@ -7890,6 +7913,22 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/**
+ * Checks if message sending has completely failed due to the timeout. Sets {@code RingMessageSendState#FAILED}
+ * if the timeout is reached.
+ *
+ * @return {@code True} if passed timeout is reached. {@code False} otherwise.
+ */
+ boolean checkTimeout() {
+ if (System.nanoTime() >= failTimeNanos) {
+ state = RingMessageSendState.FAILED;
+
+ return true;
+ }
+
+ return false;
+ }
+
+ /**
* Marks last failed node as alive.
*
* @return {@code False} if all failed nodes marked as alive or incorrect state.
@@ -7901,12 +7940,6 @@ class ServerImpl extends TcpDiscoveryImpl {
if (--failedNodes <= 0) {
failedNodes = 0;
- if (System.nanoTime() - failTimeNanos >= 0) {
- state = RingMessageSendState.FAILED;
-
- return false;
- }
-
state = RingMessageSendState.STARTING_POINT;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index 1fe3202..6f9ad5b 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -2494,11 +2494,12 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements IgniteDiscovery
LT.warn(log, "Socket write has timed out (consider increasing " +
(failureDetectionTimeoutEnabled() ?
- "'IgniteConfiguration.failureDetectionTimeout' configuration property) [" +
- "failureDetectionTimeout=" + failureDetectionTimeout() :
- "'sockTimeout' configuration property) [sockTimeout=" + sockTimeout) +
- ", rmtAddr=" + sock.getRemoteSocketAddress() + ", rmtPort=" + sock.getPort() +
- ", sockTimeout=" + sockTimeout + ']');
+ "'IgniteConfiguration.failureDetectionTimeout' and 'connRecoveryTimeout' configuration " +
+ "properties) [failureDetectionTimeout=" + failureDetectionTimeout() :
+ "'sockTimeout' and 'connRecoveryTimeout' configuration properties) [sockTimeout="
+ + sockTimeout) + ", connRecoveryTimeout=" + getConnectionRecoveryTimeout() + ", rmtAddr="
+ + sock.getRemoteSocketAddress() + ", rmtPort=" + sock.getPort() + ", sockTimeout=" + sockTimeout
+ + ']');
stats.onSocketTimeout();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
index 16fd751..fcf510f 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
@@ -1976,7 +1976,7 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
ignite1.configuration().getDiscoverySpi().failNode(coordId, null);
- assertTrue(failedLatch.await(2000, MILLISECONDS));
+ assertTrue(failedLatch.await(ignite1.configuration().getFailureDetectionTimeout(), MILLISECONDS));
assertTrue(coordSegmented.get());