You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by dg...@apache.org on 2019/01/24 09:32:03 UTC
[ignite] branch master updated: IGNITE-11016 Fix
RecoveryLastReceivedMessage(NEED_WAIT) fails with "Failed to encrypt data
(SSL engine error)"
This is an automated email from the ASF dual-hosted git repository.
dgovorukhin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 82781d1 IGNITE-11016 Fix RecoveryLastReceivedMessage(NEED_WAIT) fails with "Failed to encrypt data (SSL engine error)"
82781d1 is described below
commit 82781d1d3e6d0a90a5b377f895fed3634a670ebb
Author: Pavel Voronkin <pv...@gridgain.com>
AuthorDate: Tue Dec 25 14:01:50 2018 +0300
IGNITE-11016 Fix RecoveryLastReceivedMessage(NEED_WAIT) fails with "Failed to encrypt data (SSL engine error)"
---
.../apache/ignite/internal/util/IgniteUtils.java | 25 ++++++
.../ignite/internal/util/nio/GridNioServer.java | 42 ++++++++--
.../spi/communication/tcp/TcpCommunicationSpi.java | 92 ++++++++++++++++++----
.../discovery/tcp/IgniteClientConnectSslTest.java | 41 ++++++++++
.../spi/discovery/tcp/IgniteClientConnectTest.java | 11 +--
...gniteClientReconnectMassiveShutdownSslTest.java | 38 +++++++++
.../IgniteSpiDiscoverySelfTestSuite.java | 4 +
.../ApiParity/TcpCommunicationSpiParityTest.cs | 5 +-
8 files changed, 230 insertions(+), 28 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index d9b74fd..b9e01cc 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -75,6 +75,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileLock;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.file.DirectoryStream;
import java.nio.file.FileVisitResult;
@@ -11165,4 +11166,28 @@ public abstract class IgniteUtils {
throw new IgniteException(e);
}
}
+
+ /**
+ * Safely write buffer fully to blocking socket channel.
+ * Will throw assert if non blocking channel passed.
+ *
+ * @param sockCh WritableByteChannel.
+ * @param buf Buffer.
+ * @throws IOException IOException.
+ */
+ public static void writeFully(SocketChannel sockCh, ByteBuffer buf) throws IOException {
+ int totalWritten = 0;
+
+ assert sockCh.isBlocking() : "SocketChannel should be in blocking mode " + sockCh;
+
+ while (buf.hasRemaining()) {
+ int written = sockCh.write(buf);
+
+ if (written < 0)
+ throw new IOException("Error writing buffer to channel " +
+ "[written = " + written + ", buf " + buf + ", totalWritten = " + totalWritten + "]");
+
+ totalWritten += written;
+ }
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
index 7e3221e..794c8fc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
@@ -122,6 +122,9 @@ public class GridNioServer<T> {
/** Selection key meta key. */
private static final int WORKER_IDX_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
+ /** Meta key for pending requests to be written. */
+ private static final int REQUESTS_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
+
/** */
private static final boolean DISABLE_KEYSET_OPTIMIZATION =
IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_NO_SELECTOR_OPTS);
@@ -1373,12 +1376,18 @@ public class GridNioServer<T> {
return;
}
+ else {
+ List<SessionWriteRequest> requests = ses.removeMeta(REQUESTS_META_KEY);
+
+ if (requests != null)
+ onRequestsWritten(ses, requests);
+ }
}
ByteBuffer buf = ses.writeBuffer();
if (ses.meta(WRITE_BUF_LIMIT) != null)
- buf.limit((int)ses.meta(WRITE_BUF_LIMIT));
+ buf.limit(ses.meta(WRITE_BUF_LIMIT));
SessionWriteRequest req = ses.removeMeta(NIO_OPERATION.ordinal());
@@ -1409,8 +1418,10 @@ public class GridNioServer<T> {
Message msg;
boolean finished = false;
+ List<SessionWriteRequest> pendingRequests = new ArrayList<>(2);
+
if (req != null) {
- msg = (Message)req.message();
+ msg = (Message) req.message();
assert msg != null;
@@ -1420,7 +1431,7 @@ public class GridNioServer<T> {
finished = msg.writeTo(buf, writer);
if (finished) {
- onMessageWritten(ses, msg);
+ pendingRequests.add(req);
if (writer != null)
writer.reset();
@@ -1429,8 +1440,6 @@ public class GridNioServer<T> {
// Fill up as many messages as possible to write buffer.
while (finished) {
- req.onMessageWritten();
-
req = systemMessage(ses);
if (req == null)
@@ -1439,7 +1448,7 @@ public class GridNioServer<T> {
if (req == null)
break;
- msg = (Message)req.message();
+ msg = (Message) req.message();
assert msg != null;
@@ -1449,7 +1458,7 @@ public class GridNioServer<T> {
finished = msg.writeTo(buf, writer);
if (finished) {
- onMessageWritten(ses, msg);
+ pendingRequests.add(req);
if (writer != null)
writer.reset();
@@ -1503,13 +1512,17 @@ public class GridNioServer<T> {
if (buf.hasRemaining()) {
ses.addMeta(BUF_META_KEY, buf);
+ ses.addMeta(REQUESTS_META_KEY, pendingRequests);
+
break;
}
else {
+ onRequestsWritten(ses, pendingRequests);
+
buf = ses.writeBuffer();
if (ses.meta(WRITE_BUF_LIMIT) != null)
- buf.limit((int)ses.meta(WRITE_BUF_LIMIT));
+ buf.limit(ses.meta(WRITE_BUF_LIMIT));
}
}
}
@@ -1703,6 +1716,19 @@ public class GridNioServer<T> {
}
/**
+ * Notifies SessionWriteRequests and it's messages when requests were actually written.
+ * @param ses GridNioSession.
+ * @param requests SessionWriteRequests.
+ */
+ private void onRequestsWritten(GridSelectorNioSessionImpl ses, List<SessionWriteRequest> requests) {
+ for (SessionWriteRequest request : requests) {
+ request.onMessageWritten();
+
+ onMessageWritten(ses, (Message)request.message());
+ }
+ }
+
+ /**
* Handle message written event.
*
* @param ses Session.
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 044ffef..2892721 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
@@ -360,6 +360,12 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
/** Default connections per node. */
public static final int DFLT_CONN_PER_NODE = 1;
+ /** Default start delay in case of NEED_WAIT received on handshake, millis. */
+ public static final long DFLT_NEED_WAIT_DELAY = 200;
+
+ /** Default max delay in case of NEED_WAIT received on handshake, millis. */
+ public static final long DFLT_MAX_NEED_WAIT_DELAY = 60_000;
+
/** No-op runnable. */
private static final IgniteRunnable NOOP = new IgniteRunnable() {
@Override public void run() {
@@ -1174,6 +1180,12 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
/** Count of selectors to use in TCP server. */
private int selectorsCnt = DFLT_SELECTORS_CNT;
+ /** Delay on subsequent retry if NEED_WAIT received. */
+ private long needWaitDelay = DFLT_NEED_WAIT_DELAY;
+
+ /** Max value of needWaitDelay. */
+ private long maxNeedWaitDelay = DFLT_MAX_NEED_WAIT_DELAY;
+
/**
* Defines how many non-blocking {@code selector.selectNow()} should be made before
* falling into {@code selector.select(long)} in NIO server. Long value. Default is {@code 0}.
@@ -1874,6 +1886,56 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
return this;
}
+ /**
+ * Get initial needWaitDelay.
+ * @return Initial needWaitDelay.
+ */
+ public long getNeedWaitDelay() {
+ return needWaitDelay;
+ }
+
+ /**
+ * Setting to set start needWaitDelay.
+ * <p>
+ * Configures delay on subsequent reconnect if NEED_WAIT received.
+ * <p>
+ * If not provided, default value is {@link #DFLT_NEED_WAIT_DELAY}.
+ *
+ * @param needWaitDelay Delay between subsequent retries on NEED_WAIT.
+ * @return {@code this} for chaining.
+ */
+ @IgniteSpiConfiguration(optional = true)
+ public TcpCommunicationSpi setNeedWaitDelay(long needWaitDelay) {
+ this.needWaitDelay = needWaitDelay;
+
+ return this;
+ }
+
+ /**
+ * Get max needWaitDelay.
+ * @return Max needWaitDelay.
+ */
+ public long getMaxNeedWaitDelay() {
+ return maxNeedWaitDelay;
+ }
+
+ /**
+ * Setting to set start needWaitDelay.
+ * <p>
+ * Configures max delay on subsequent reconnect if NEED_WAIT received.
+ * <p>
+ * If not provided, default value is {@link #DFLT_MAX_NEED_WAIT_DELAY}.
+ *
+ * @param maxNeedWaitDelay Max delay between subsequent retries on NEED_WAIT.
+ * @return {@code this} for chaining.
+ */
+ @IgniteSpiConfiguration(optional = true)
+ public TcpCommunicationSpi setMaxNeedWaitDelay(long maxNeedWaitDelay) {
+ this.maxNeedWaitDelay = maxNeedWaitDelay;
+
+ return this;
+ }
+
/** {@inheritDoc} */
@Override public void setListener(CommunicationListener<Message> lsnr) {
this.lsnr = lsnr;
@@ -3272,7 +3334,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
IgniteSpiOperationTimeoutHelper timeoutHelper = new IgniteSpiOperationTimeoutHelper(this,
!node.isClient());
- int lastWaitingTimeout = 1;
+ long needWaitDelay0 = needWaitDelay;
while (client == null) { // Reconnection on handshake timeout.
if (stopping)
@@ -3355,12 +3417,10 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
sslMeta,
handshakeConnIdx);
- if (rcvCnt == ALREADY_CONNECTED) {
+ if (rcvCnt == ALREADY_CONNECTED)
return null;
- }
- else if (rcvCnt == NODE_STOPPING) {
+ else if (rcvCnt == NODE_STOPPING)
throw new ClusterTopologyCheckedException("Remote node started stop procedure: " + node.id());
- }
else if (rcvCnt == NEED_WAIT) {
needWait = true;
@@ -3387,10 +3447,14 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
recoveryDesc.release();
if (needWait) {
- if (lastWaitingTimeout < 60000)
- lastWaitingTimeout *= 2;
+ if (needWaitDelay0 < maxNeedWaitDelay)
+ needWaitDelay0 *= 2;
+
+ if (log.isDebugEnabled())
+ log.debug("NEED_WAIT received, reconnect after delay [node = "
+ + node + ", delay = " + needWaitDelay0 + "ms]");
- U.sleep(lastWaitingTimeout);
+ U.sleep(needWaitDelay0);
}
}
}
@@ -3742,10 +3806,10 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
if (isSslEnabled()) {
assert sslHnd != null;
- ch.write(sslHnd.encrypt(ByteBuffer.wrap(U.IGNITE_HEADER)));
+ U.writeFully(ch, sslHnd.encrypt(ByteBuffer.wrap(U.IGNITE_HEADER)));
}
else
- ch.write(ByteBuffer.wrap(U.IGNITE_HEADER));
+ U.writeFully(ch, ByteBuffer.wrap(U.IGNITE_HEADER));
ClusterNode locNode = getLocalNode();
@@ -3789,19 +3853,19 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
if (isSslEnabled()) {
assert sslHnd != null;
- ch.write(sslHnd.encrypt(buf));
+ U.writeFully(ch, sslHnd.encrypt(buf));
}
else
- ch.write(buf);
+ U.writeFully(ch, buf);
}
else {
if (isSslEnabled()) {
assert sslHnd != null;
- ch.write(sslHnd.encrypt(ByteBuffer.wrap(NodeIdMessage.nodeIdBytesWithType(safeLocalNodeId()))));
+ U.writeFully(ch, sslHnd.encrypt(ByteBuffer.wrap(NodeIdMessage.nodeIdBytesWithType(safeLocalNodeId()))));
}
else
- ch.write(ByteBuffer.wrap(NodeIdMessage.nodeIdBytesWithType(safeLocalNodeId())));
+ U.writeFully(ch, ByteBuffer.wrap(NodeIdMessage.nodeIdBytesWithType(safeLocalNodeId())));
}
if (recovery != null) {
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientConnectSslTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientConnectSslTest.java
new file mode 100644
index 0000000..d6d27f1
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientConnectSslTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.discovery.tcp;
+
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * We emulate that client receive message about joining to topology earlier than some server nodes in topology.
+ * And make this client connect to such servers.
+ * To emulate this we connect client to second node in topology and pause sending message about joining finishing to
+ * third node.
+ */
+@RunWith(JUnit4.class)
+public class IgniteClientConnectSslTest extends IgniteClientConnectTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setSslContextFactory(GridTestUtils.sslFactory());
+
+ return cfg;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientConnectTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientConnectTest.java
index 784078f..b591458 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientConnectTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientConnectTest.java
@@ -55,14 +55,14 @@ public class IgniteClientConnectTest extends GridCommonAbstractTest {
/** Start client flag. */
private final AtomicBoolean clientJustStarted = new AtomicBoolean(false);
-
+
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
TestTcpDiscoverySpi disco = new TestTcpDiscoverySpi();
- if (igniteInstanceName.equals("client")) {
+ if ("client".equals(igniteInstanceName)) {
TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder();
ipFinder.registerAddresses(Collections.singleton(new InetSocketAddress(InetAddress.getLoopbackAddress(), 47501)));
@@ -74,7 +74,9 @@ public class IgniteClientConnectTest extends GridCommonAbstractTest {
disco.setJoinTimeout(2 * 60_000);
disco.setSocketTimeout(1000);
- disco.setNetworkTimeout(2000);
+ disco.setNetworkTimeout(6_000);
+
+ cfg.setNetworkSendRetryCount(1);
cfg.setDiscoverySpi(disco);
@@ -121,7 +123,6 @@ public class IgniteClientConnectTest extends GridCommonAbstractTest {
latch.countDown();
- System.err.println("GET ALL");
client.cache(DEFAULT_CACHE_NAME).getAll(keys);
}
@@ -142,7 +143,7 @@ public class IgniteClientConnectTest extends GridCommonAbstractTest {
try {
latch.await();
- Thread.sleep(3000);
+ Thread.sleep(5_000);
} catch (InterruptedException e) {
e.printStackTrace();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientReconnectMassiveShutdownSslTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientReconnectMassiveShutdownSslTest.java
new file mode 100644
index 0000000..2f396c2
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteClientReconnectMassiveShutdownSslTest.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.discovery.tcp;
+
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Client reconnect test in multi threaded mode while cache operations are in progress.
+ */
+@RunWith(JUnit4.class)
+public class IgniteClientReconnectMassiveShutdownSslTest extends IgniteClientReconnectMassiveShutdownTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setSslContextFactory(GridTestUtils.sslFactory());
+
+ return cfg;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java
index 18776b4..d37baa6 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java
@@ -24,7 +24,9 @@ import org.apache.ignite.spi.discovery.FilterDataForClientNodeDiscoveryTest;
import org.apache.ignite.spi.discovery.IgniteDiscoveryCacheReuseSelfTest;
import org.apache.ignite.spi.discovery.LongClientConnectToClusterTest;
import org.apache.ignite.spi.discovery.tcp.DiscoveryUnmarshalVulnerabilityTest;
+import org.apache.ignite.spi.discovery.tcp.IgniteClientConnectSslTest;
import org.apache.ignite.spi.discovery.tcp.IgniteClientConnectTest;
+import org.apache.ignite.spi.discovery.tcp.IgniteClientReconnectMassiveShutdownSslTest;
import org.apache.ignite.spi.discovery.tcp.IgniteClientReconnectMassiveShutdownTest;
import org.apache.ignite.spi.discovery.tcp.TcpClientDiscoveryMarshallerCheckSelfTest;
import org.apache.ignite.spi.discovery.tcp.TcpClientDiscoverySpiCoordinatorChangeTest;
@@ -121,7 +123,9 @@ import static org.apache.ignite.IgniteSystemProperties.IGNITE_OVERRIDE_MCAST_GRP
// Client connect.
IgniteClientConnectTest.class,
+ IgniteClientConnectSslTest.class,
IgniteClientReconnectMassiveShutdownTest.class,
+ IgniteClientReconnectMassiveShutdownSslTest.class,
TcpDiscoveryClientSuspensionSelfTest.class,
// SSL.
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/TcpCommunicationSpiParityTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/TcpCommunicationSpiParityTest.cs
index be8bd11..5d301cc 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/TcpCommunicationSpiParityTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/TcpCommunicationSpiParityTest.cs
@@ -65,7 +65,10 @@ namespace Apache.Ignite.Core.Tests.ApiParity
};
/** Properties that are missing on .NET side. */
- private static readonly string[] MissingProperties = {};
+ private static readonly string[] MissingProperties = {
+ "NeedWaitDelay", // IGNITE-11026
+ "MaxNeedWaitDelay" // IGNITE-11026
+ };
/// <summary>
/// Tests the cache configuration parity.