You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2017/03/30 14:38:15 UTC
[1/9] ignite git commit: IGNITE-4617: Fix for BinaryFieldMetadata
[Forced Update!]
Repository: ignite
Updated Branches:
refs/heads/ignite-4003 ffa5f8266 -> 04aca2e62 (forced update)
IGNITE-4617: Fix for BinaryFieldMetadata
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/caa16b33
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/caa16b33
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/caa16b33
Branch: refs/heads/ignite-4003
Commit: caa16b33ae236efb8b4319ba104e73f87e32f895
Parents: 348b9ef
Author: Igor Sapego <is...@gridgain.com>
Authored: Wed Mar 29 16:01:37 2017 +0300
Committer: Igor Sapego <is...@gridgain.com>
Committed: Wed Mar 29 16:01:37 2017 +0300
----------------------------------------------------------------------
.../java/org/apache/ignite/internal/binary/BinaryMetadata.java | 4 ++--
modules/core/src/main/resources/META-INF/classnames.properties | 1 +
2 files changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/caa16b33/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
index a2589bf..d1c79f3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
@@ -125,9 +125,9 @@ public class BinaryMetadata implements Externalizable {
* @return Field type name.
*/
@Nullable public String fieldTypeName(String fieldName) {
- Integer typeId = fields != null ? fields.get(fieldName).typeId() : null;
+ BinaryFieldMetadata meta = fields != null ? fields.get(fieldName) : null;
- return typeId != null ? BinaryUtils.fieldTypeName(typeId) : null;
+ return meta != null ? BinaryUtils.fieldTypeName(meta.typeId()) : null;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/caa16b33/modules/core/src/main/resources/META-INF/classnames.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index 698d358..c5f4b92 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -235,6 +235,7 @@ org.apache.ignite.internal.IgniteServicesImpl
org.apache.ignite.internal.IgnitionEx$IgniteNamedInstance$1
org.apache.ignite.internal.NodeStoppingException
org.apache.ignite.internal.binary.BinaryEnumObjectImpl
+org.apache.ignite.internal.binary.BinaryFieldMetadata
org.apache.ignite.internal.binary.BinaryMetadata
org.apache.ignite.internal.binary.BinaryObjectEx
org.apache.ignite.internal.binary.BinaryObjectExImpl
[2/9] ignite git commit: IGNITE-4141 - JDBC driver should always set
withKeepBinary flag when querying cache. This fixes #1617.
Posted by ag...@apache.org.
IGNITE-4141 - JDBC driver should always set withKeepBinary flag when querying cache. This fixes #1617.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8dd88d81
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8dd88d81
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8dd88d81
Branch: refs/heads/ignite-4003
Commit: 8dd88d81cf2174fbde920ad00c892995cf913711
Parents: caa16b3
Author: Evgenii Zhuravlev <ez...@gridgain.com>
Authored: Wed Mar 29 15:28:20 2017 -0700
Committer: Valentin Kulichenko <va...@gmail.com>
Committed: Wed Mar 29 15:28:20 2017 -0700
----------------------------------------------------------------------
.../jdbc/AbstractJdbcPojoQuerySelfTest.java | 169 +++++++++++++++++++
.../jdbc/JdbcPojoLegacyQuerySelfTest.java | 44 +++++
.../ignite/jdbc/JdbcPojoQuerySelfTest.java | 56 ++++++
.../jdbc/suite/IgniteJdbcDriverTestSuite.java | 4 +
.../ignite/internal/jdbc2/JdbcQueryTask.java | 2 +-
.../ignite/internal/jdbc2/JdbcQueryTaskV2.java | 2 +-
.../query/jdbc/GridCacheQueryJdbcTask.java | 4 +-
7 files changed, 277 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/8dd88d81/modules/clients/src/test/java/org/apache/ignite/jdbc/AbstractJdbcPojoQuerySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/AbstractJdbcPojoQuerySelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/AbstractJdbcPojoQuerySelfTest.java
new file mode 100644
index 0000000..a150574
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/AbstractJdbcPojoQuerySelfTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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.jdbc;
+
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Collections;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.config.GridTestProperties;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Test for Jdbc driver query without class on client
+ */
+public abstract class AbstractJdbcPojoQuerySelfTest extends GridCommonAbstractTest {
+ /** IP finder. */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** TestObject class name. */
+ protected static final String TEST_OBJECT = "org.apache.ignite.internal.JdbcTestObject";
+
+ /** TestObject class name. */
+ protected static final String TEST_OBJECT_2 = "org.apache.ignite.internal.JdbcTestObject2";
+
+ /** Statement. */
+ protected Statement stmt;
+
+ /** */
+ private String marshallerBackup = GridTestProperties.getProperty(GridTestProperties.MARSH_CLASS_NAME);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ GridTestProperties.setProperty(GridTestProperties.MARSH_CLASS_NAME, BinaryMarshaller.class.getName());
+
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration<?,?> cache = defaultCacheConfiguration();
+
+ cache.setWriteSynchronizationMode(FULL_SYNC);
+ cache.setAtomicityMode(TRANSACTIONAL);
+
+ TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+ disco.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(disco);
+ cfg.setConnectorConfiguration(new ConnectorConfiguration());
+
+ QueryEntity queryEntity = new QueryEntity();
+ queryEntity.setKeyType("java.lang.String");
+ queryEntity.setValueType("org.apache.ignite.internal.JdbcTestObject");
+ queryEntity.addQueryField("id", "java.lang.Integer", null);
+ queryEntity.addQueryField("testObject", "org.apache.ignite.internal.JdbcTestObject2", null);
+ queryEntity.setIndexes(Collections.singletonList(new QueryIndex("id")));
+
+ cache.setQueryEntities(Collections.singletonList(queryEntity));
+
+ cfg.setCacheConfiguration(cache);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ Ignite ignite = startGrid(0);
+
+ BinaryObjectBuilder builder = ignite.binary().builder(TEST_OBJECT);
+ BinaryObjectBuilder builder2 = ignite.binary().builder(TEST_OBJECT_2);
+
+ builder2.setField("id", 1);
+ builder2.setField("boolVal", true);
+
+ BinaryObject testObject = builder2.build();
+
+ builder.setField("id", 1);
+ builder.setField("testObject", testObject);
+
+ BinaryObject binObj = builder.build();
+
+ IgniteCache<String, BinaryObject> cache = grid(0).cache(null);
+
+ cache.put("0", binObj);
+
+ Class.forName("org.apache.ignite.IgniteJdbcDriver");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ GridTestProperties.setProperty(GridTestProperties.MARSH_CLASS_NAME, marshallerBackup);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ stmt = DriverManager.getConnection(getURL()).createStatement();
+
+ assertNotNull(stmt);
+ assertFalse(stmt.isClosed());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ if (stmt != null) {
+ stmt.getConnection().close();
+ stmt.close();
+
+ assertTrue(stmt.isClosed());
+ }
+ }
+
+ /**
+ * @param rs Result set.
+ * @throws Exception In case of error.
+ */
+ protected void assertResultSet(ResultSet rs) throws Exception {
+ assertNotNull(rs);
+
+ int cnt = 0;
+
+ while (rs.next()) {
+ assertNotNull(rs.getString("id"));
+ assertNotNull(rs.getString("testObject"));
+
+ assertTrue(rs.getObject("testObject").toString().contains("id=1"));
+ assertTrue(rs.getObject("testObject").toString().contains("boolVal=true"));
+
+ cnt++;
+ }
+
+ assertEquals(1, cnt);
+ }
+
+ /**
+ * @return URL.
+ */
+ protected abstract String getURL();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8dd88d81/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoLegacyQuerySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoLegacyQuerySelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoLegacyQuerySelfTest.java
new file mode 100644
index 0000000..4fa7ba5
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoLegacyQuerySelfTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.jdbc;
+
+import java.sql.ResultSet;
+
+/**
+ * Test for Jdbc driver query without class on client
+ */
+public class JdbcPojoLegacyQuerySelfTest extends AbstractJdbcPojoQuerySelfTest {
+ /** URL. */
+ private static final String URL = "jdbc:ignite://127.0.0.1/";
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJdbcQuery() throws Exception {
+ stmt.execute("select * from JdbcTestObject");
+
+ ResultSet rs = stmt.getResultSet();
+
+ assertResultSet(rs);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected String getURL() {
+ return URL;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8dd88d81/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoQuerySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoQuerySelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoQuerySelfTest.java
new file mode 100644
index 0000000..c2af8a1
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/JdbcPojoQuerySelfTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.jdbc;
+
+import java.sql.ResultSet;
+
+import static org.apache.ignite.IgniteJdbcDriver.CFG_URL_PREFIX;
+
+/**
+ * Test for Jdbc driver query without class on client
+ */
+public class JdbcPojoQuerySelfTest extends AbstractJdbcPojoQuerySelfTest {
+ /** URL. */
+ private static final String URL = CFG_URL_PREFIX + "modules/clients/src/test/config/jdbc-bin-config.xml";
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJdbcQueryTask2() throws Exception {
+ stmt.execute("select * from JdbcTestObject");
+
+ ResultSet rs = stmt.getResultSet();
+
+ assertResultSet(rs);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJdbcQueryTask1() throws Exception {
+ ResultSet rs = stmt.executeQuery("select * from JdbcTestObject");
+
+ assertResultSet(rs);
+
+ }
+
+ /** {@inheritDoc} */
+ @Override protected String getURL() {
+ return URL;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8dd88d81/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
index 85e7d90..2489de9 100644
--- a/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/suite/IgniteJdbcDriverTestSuite.java
@@ -24,6 +24,8 @@ import org.apache.ignite.jdbc.JdbcEmptyCacheSelfTest;
import org.apache.ignite.jdbc.JdbcLocalCachesSelfTest;
import org.apache.ignite.jdbc.JdbcMetadataSelfTest;
import org.apache.ignite.jdbc.JdbcNoDefaultCacheTest;
+import org.apache.ignite.jdbc.JdbcPojoLegacyQuerySelfTest;
+import org.apache.ignite.jdbc.JdbcPojoQuerySelfTest;
import org.apache.ignite.jdbc.JdbcPreparedStatementSelfTest;
import org.apache.ignite.jdbc.JdbcResultSetSelfTest;
import org.apache.ignite.jdbc.JdbcStatementSelfTest;
@@ -49,6 +51,8 @@ public class IgniteJdbcDriverTestSuite extends TestSuite {
suite.addTest(new TestSuite(JdbcEmptyCacheSelfTest.class));
suite.addTest(new TestSuite(JdbcLocalCachesSelfTest.class));
suite.addTest(new TestSuite(JdbcNoDefaultCacheTest.class));
+ suite.addTest(new TestSuite(JdbcPojoQuerySelfTest.class));
+ suite.addTest(new TestSuite(JdbcPojoLegacyQuerySelfTest.class));
// Ignite client node based driver tests
suite.addTest(new TestSuite(org.apache.ignite.internal.jdbc2.JdbcConnectionSelfTest.class));
http://git-wip-us.apache.org/repos/asf/ignite/blob/8dd88d81/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
index 0b23f9b..bd6b0f2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTask.java
@@ -158,7 +158,7 @@ class JdbcQueryTask implements IgniteCallable<JdbcQueryTask.QueryResult> {
qry.setCollocated(collocatedQry);
qry.setDistributedJoins(distributedJoins);
- QueryCursor<List<?>> qryCursor = cache.query(qry);
+ QueryCursor<List<?>> qryCursor = cache.withKeepBinary().query(qry);
Collection<GridQueryFieldMetadata> meta = ((QueryCursorImpl<List<?>>)qryCursor).fieldsMeta();
http://git-wip-us.apache.org/repos/asf/ignite/blob/8dd88d81/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTaskV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTaskV2.java b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTaskV2.java
index 9093d15..61f152d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTaskV2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcQueryTaskV2.java
@@ -161,7 +161,7 @@ class JdbcQueryTaskV2 implements IgniteCallable<JdbcQueryTaskV2.QueryResult> {
qry.setCollocated(collocatedQry);
qry.setDistributedJoins(distributedJoins);
- QueryCursorImpl<List<?>> qryCursor = (QueryCursorImpl<List<?>>)cache.query(qry);
+ QueryCursorImpl<List<?>> qryCursor = (QueryCursorImpl<List<?>>)cache.withKeepBinary().query(qry);
if (isQry == null)
isQry = qryCursor.isQuery();
http://git-wip-us.apache.org/repos/asf/ignite/blob/8dd88d81/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/jdbc/GridCacheQueryJdbcTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/jdbc/GridCacheQueryJdbcTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/jdbc/GridCacheQueryJdbcTask.java
index ca08ead..4ae8a8a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/jdbc/GridCacheQueryJdbcTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/jdbc/GridCacheQueryJdbcTask.java
@@ -268,7 +268,7 @@ public class GridCacheQueryJdbcTask extends ComputeTaskAdapter<byte[], byte[]> {
qry.setPageSize(pageSize);
- QueryCursor<List<?>> cursor = cache.query(qry);
+ QueryCursor<List<?>> cursor = cache.withKeepBinary().query(qry);
Collection<GridQueryFieldMetadata> meta = ((QueryCursorImpl<List<?>>)cursor).fieldsMeta();
@@ -453,4 +453,4 @@ public class GridCacheQueryJdbcTask extends ComputeTaskAdapter<byte[], byte[]> {
return iter;
}
}
-}
\ No newline at end of file
+}
[7/9] ignite git commit: ignite-4003 Async outgoing connections for
communication SPI
Posted by ag...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheMessageWriteTimeoutTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheMessageWriteTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheMessageWriteTimeoutTest.java
index b3b5d1a..3ba319b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheMessageWriteTimeoutTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheMessageWriteTimeoutTest.java
@@ -50,8 +50,8 @@ public class IgniteCacheMessageWriteTimeoutTest extends GridCommonAbstractTest {
// Try provoke connection close on socket writeTimeout.
commSpi.setSharedMemoryPort(-1);
commSpi.setMessageQueueLimit(10);
- commSpi.setSocketReceiveBuffer(40);
- commSpi.setSocketSendBuffer(40);
+ commSpi.setSocketReceiveBuffer(64);
+ commSpi.setSocketSendBuffer(64);
commSpi.setSocketWriteTimeout(100);
commSpi.setUnacknowledgedMessagesBufferSize(1000);
commSpi.setConnectTimeout(10_000);
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java
index 4dbb7ce..e623467 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java
@@ -678,7 +678,7 @@ public class GridNioSelfTest extends GridCommonAbstractTest {
try {
SocketChannel ch = SocketChannel.open(new InetSocketAddress(U.getLocalHost(), srvr2.port()));
- GridNioFuture<GridNioSession> fut = srvr1.createSession(ch, null);
+ GridNioFuture<GridNioSession> fut = srvr1.createSession(ch, null, false, null);
ses = fut.get();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/GridTcpSpiForwardingSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/GridTcpSpiForwardingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/GridTcpSpiForwardingSelfTest.java
index 1e25003..bee63b3 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/GridTcpSpiForwardingSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/GridTcpSpiForwardingSelfTest.java
@@ -30,6 +30,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.AddressResolver;
import org.apache.ignite.configuration.BasicAddressResolver;
import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.util.nio.GridCommunicationClient;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteCallable;
@@ -111,7 +112,7 @@ public class GridTcpSpiForwardingSelfTest extends GridCommonAbstractTest {
cfg.setConnectorConfiguration(null);
TcpCommunicationSpi commSpi = new TcpCommunicationSpi() {
- @Override protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx)
+ @Override protected IgniteInternalFuture<GridCommunicationClient> createTcpClient(ClusterNode node, int connIdx)
throws IgniteCheckedException {
Map<String, Object> attrs = new HashMap<>(node.attributes());
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
index 88276c2..07edc86 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
@@ -30,6 +30,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.util.typedef.CO;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -39,6 +40,7 @@ import org.apache.ignite.spi.IgniteSpiAdapter;
import org.apache.ignite.testframework.GridSpiTestContext;
import org.apache.ignite.testframework.GridTestNode;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.IgniteMock;
import org.apache.ignite.testframework.junits.IgniteTestResources;
import org.apache.ignite.testframework.junits.spi.GridSpiAbstractTest;
@@ -70,6 +72,9 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
private static final Object mux = new Object();
/** */
+ private static GridTimeoutProcessor timeoutProcessor;
+
+ /** */
protected boolean useSsl = false;
/**
@@ -289,6 +294,12 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
Map<ClusterNode, GridSpiTestContext> ctxs = new HashMap<>();
+ timeoutProcessor = new GridTimeoutProcessor(new GridTestKernalContext(log));
+
+ timeoutProcessor.start();
+
+ timeoutProcessor.onKernalStart();
+
for (int i = 0; i < getSpiCount(); i++) {
CommunicationSpi<Message> spi = getSpi(i);
@@ -298,18 +309,20 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
GridTestNode node = new GridTestNode(rsrcs.getNodeId());
- node.order(i);
-
GridSpiTestContext ctx = initSpiContext();
ctx.setLocalNode(node);
+ ctx.timeoutProcessor(timeoutProcessor);
+
info(">>> Initialized context: nodeId=" + ctx.localNode().id());
spiRsrcs.add(rsrcs);
rsrcs.inject(spi);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
+
if (useSsl) {
IgniteMock ignite = GridTestUtils.getFieldValue(spi, IgniteSpiAdapter.class, "ignite");
@@ -324,6 +337,8 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
node.setAttributes(spi.getNodeAttributes());
node.setAttribute(ATTR_MACS, F.concat(U.allLocalMACs(), ", "));
+ node.order(i + 1);
+
nodes.add(node);
spi.spiStart(getTestIgniteInstanceName() + (i + 1));
@@ -346,6 +361,14 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
/** {@inheritDoc} */
@Override protected void afterTestsStopped() throws Exception {
+ if (timeoutProcessor != null) {
+ timeoutProcessor.onKernalStop(true);
+
+ timeoutProcessor.stop(true);
+
+ timeoutProcessor = null;
+ }
+
for (CommunicationSpi<Message> spi : spis.values()) {
spi.onContextDestroyed();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
index 78bf869..39ecd8e 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
@@ -36,7 +36,9 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.nio.GridCommunicationClient;
import org.apache.ignite.internal.util.nio.GridNioServer;
@@ -51,6 +53,7 @@ import org.apache.ignite.spi.communication.GridTestMessage;
import org.apache.ignite.testframework.GridSpiTestContext;
import org.apache.ignite.testframework.GridTestNode;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.IgniteMock;
import org.apache.ignite.testframework.junits.IgniteTestResources;
import org.apache.ignite.testframework.junits.spi.GridSpiAbstractTest;
@@ -79,6 +82,9 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
protected static final List<ClusterNode> nodes = new ArrayList<>();
/** */
+ private static GridTimeoutProcessor timeoutProcessor;
+
+ /** */
private static int port = 60_000;
/** Use ssl. */
@@ -407,27 +413,37 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
Map<ClusterNode, GridSpiTestContext> ctxs = new HashMap<>();
+ timeoutProcessor = new GridTimeoutProcessor(new GridTestKernalContext(log));
+
+ timeoutProcessor.start();
+
+ timeoutProcessor.onKernalStart();
+
for (int i = 0; i < SPI_CNT; i++) {
CommunicationSpi<Message> spi = createSpi();
- GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
-
IgniteTestResources rsrcs = new IgniteTestResources();
GridTestNode node = new GridTestNode(rsrcs.getNodeId());
+ node.setAttribute(IgniteNodeAttributes.ATTR_CLIENT_MODE, false);
+
node.order(i + 1);
GridSpiTestContext ctx = initSpiContext();
ctx.setLocalNode(node);
+ ctx.timeoutProcessor(timeoutProcessor);
+
info(">>> Initialized context: nodeId=" + ctx.localNode().id());
spiRsrcs.add(rsrcs);
rsrcs.inject(spi);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
+
if (useSsl) {
IgniteMock ignite = GridTestUtils.getFieldValue(spi, IgniteSpiAdapter.class, "ignite");
@@ -494,6 +510,14 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
* @throws Exception If failed.
*/
private void stopSpis() throws Exception {
+ if (timeoutProcessor != null) {
+ timeoutProcessor.onKernalStop(true);
+
+ timeoutProcessor.stop(true);
+
+ timeoutProcessor = null;
+ }
+
for (CommunicationSpi<Message> spi : spis) {
spi.onContextDestroyed();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
index feaae11..f87ff09 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
@@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.nio.GridNioRecoveryDescriptor;
import org.apache.ignite.internal.util.nio.GridNioServer;
@@ -44,6 +45,7 @@ import org.apache.ignite.spi.communication.GridTestMessage;
import org.apache.ignite.testframework.GridSpiTestContext;
import org.apache.ignite.testframework.GridTestNode;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.IgniteTestResources;
import org.apache.ignite.testframework.junits.spi.GridSpiAbstractTest;
import org.apache.ignite.testframework.junits.spi.GridSpiTest;
@@ -64,11 +66,11 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
protected static final List<ClusterNode> nodes = new ArrayList<>();
/** */
+ private static GridTimeoutProcessor timeoutProcessor;
+
+ /** */
private static final int SPI_CNT = 2;
- /**
- *
- */
static {
GridIoMessageFactory.registerCustom(GridTestMessage.DIRECT_TYPE, new CO<Message>() {
@Override public Message apply() {
@@ -159,6 +161,8 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
spi1.sendMessage(node0, new GridTestMessage(node1.id(), ++msgId, 0));
}
+ U.sleep(500);
+
expMsgs += msgPerIter;
final long totAcked0 = totAcked;
@@ -166,9 +170,14 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
for (TcpCommunicationSpi spi : spis) {
GridNioServer srv = U.field(spi, "nioSrvr");
- Collection<? extends GridNioSession> sessions = GridTestUtils.getFieldValue(srv, "sessions");
+ final Collection<? extends GridNioSession> sessions = GridTestUtils.getFieldValue(srv, "sessions");
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return !sessions.isEmpty();
+ }
+ }, 5_000);
- assertFalse(sessions.isEmpty());
boolean found = false;
@@ -268,21 +277,21 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
ClusterNode node0 = nodes.get(0);
ClusterNode node1 = nodes.get(1);
- final GridNioServer srv1 = U.field(spi1, "nioSrvr");
-
int msgId = 0;
// Send message to establish connection.
spi0.sendMessage(node1, new GridTestMessage(node0.id(), ++msgId, 0));
+ U.sleep(1000);
+
// Prevent node1 from send
- GridTestUtils.setFieldValue(srv1, "skipWrite", true);
+ GridTestUtils.setFieldValue(spi1, "skipAck", true);
final GridNioSession ses0 = communicationSession(spi0);
int sentMsgs = 1;
- for (int i = 0; i < 150; i++) {
+ for (int i = 0; i < 1280; i++) {
try {
spi0.sendMessage(node1, new GridTestMessage(node0.id(), ++msgId, 0));
@@ -304,7 +313,7 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
assertTrue("Failed to wait for session close", ses0.closeTime() != 0);
- GridTestUtils.setFieldValue(srv1, "skipWrite", false);
+ GridTestUtils.setFieldValue(spi1, "skipAck", false);
for (int i = 0; i < 100; i++)
spi0.sendMessage(node1, new GridTestMessage(node0.id(), ++msgId, 0));
@@ -379,11 +388,15 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
Map<ClusterNode, GridSpiTestContext> ctxs = new HashMap<>();
+ timeoutProcessor = new GridTimeoutProcessor(new GridTestKernalContext(log));
+
+ timeoutProcessor.start();
+
+ timeoutProcessor.onKernalStart();
+
for (int i = 0; i < SPI_CNT; i++) {
TcpCommunicationSpi spi = getSpi(ackCnt, idleTimeout, queueLimit);
- GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
-
IgniteTestResources rsrcs = new IgniteTestResources();
GridTestNode node = new GridTestNode(rsrcs.getNodeId());
@@ -392,14 +405,20 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
ctx.setLocalNode(node);
+ ctx.timeoutProcessor(timeoutProcessor);
+
spiRsrcs.add(rsrcs);
rsrcs.inject(spi);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
+
spi.setListener(new TestListener());
node.setAttributes(spi.getNodeAttributes());
+ node.order(i);
+
nodes.add(node);
spi.spiStart(getTestIgniteInstanceName() + (i + 1));
@@ -455,6 +474,14 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
* @throws Exception If failed.
*/
private void stopSpis() throws Exception {
+ if (timeoutProcessor != null) {
+ timeoutProcessor.onKernalStop(true);
+
+ timeoutProcessor.stop(true);
+
+ timeoutProcessor = null;
+ }
+
for (CommunicationSpi<Message> spi : spis) {
spi.onContextDestroyed();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
index 2a043ee..46d2d1d 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
@@ -32,6 +32,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.nio.GridNioServer;
import org.apache.ignite.internal.util.nio.GridNioSession;
@@ -47,6 +48,7 @@ import org.apache.ignite.spi.communication.GridTestMessage;
import org.apache.ignite.testframework.GridSpiTestContext;
import org.apache.ignite.testframework.GridTestNode;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.IgniteMock;
import org.apache.ignite.testframework.junits.IgniteTestResources;
import org.apache.ignite.testframework.junits.spi.GridSpiAbstractTest;
@@ -80,6 +82,9 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
/** Use ssl. */
protected boolean useSsl;
+ /** */
+ private static GridTimeoutProcessor timeoutProcessor;
+
/**
*
*/
@@ -115,7 +120,7 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
/** {@inheritDoc} */
@Override public void onMessage(UUID nodeId, Message msg, IgniteRunnable msgC) {
- // info("Test listener received message: " + msg);
+ //info("Test listener received message: " + msg);
assertTrue("Unexpected message: " + msg, msg instanceof GridTestMessage);
@@ -186,7 +191,7 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
* @return Timeout.
*/
protected long awaitForSocketWriteTimeout() {
- return 8000;
+ return 20000;
}
/**
@@ -298,6 +303,12 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
// Send message to establish connection.
spi0.sendMessage(node1, new GridTestMessage(node0.id(), msgId.incrementAndGet(), 0));
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return lsnr1.rcvCnt.get() >= 1;
+ }
+ }, 1000);
+
final AtomicInteger sentCnt = new AtomicInteger(1);
int errCnt = 0;
@@ -413,6 +424,12 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
// Send message to establish connection.
spi0.sendMessage(node1, new GridTestMessage(node0.id(), msgId.incrementAndGet(), 0));
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return lsnr1.rcvCnt.get() >= 1;
+ }
+ }, 1000);
+
expCnt1.incrementAndGet();
int errCnt = 0;
@@ -451,7 +468,7 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
ses1.resumeReads().get();
}
catch (IgniteCheckedException ignore) {
- // Can fail is ses1 was closed.
+ // Can fail if ses1 was closed.
}
// Wait when session is closed, then try to open new connection from node1.
@@ -534,6 +551,12 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
// Send message to establish connection.
spi0.sendMessage(node1, new GridTestMessage(node0.id(), msgId.incrementAndGet(), 0));
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return lsnr1.rcvCnt.get() >= 1;
+ }
+ }, 1000);
+
final AtomicInteger sentCnt = new AtomicInteger(1);
int errCnt = 0;
@@ -686,11 +709,15 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
Map<ClusterNode, GridSpiTestContext> ctxs = new HashMap<>();
+ timeoutProcessor = new GridTimeoutProcessor(new GridTestKernalContext(log));
+
+ timeoutProcessor.start();
+
+ timeoutProcessor.onKernalStart();
+
for (int i = 0; i < SPI_CNT; i++) {
TcpCommunicationSpi spi = getSpi(i);
- GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
-
IgniteTestResources rsrcs = new IgniteTestResources();
GridTestNode node = new GridTestNode(rsrcs.getNodeId());
@@ -701,10 +728,14 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
ctx.setLocalNode(node);
+ ctx.timeoutProcessor(timeoutProcessor);
+
spiRsrcs.add(rsrcs);
rsrcs.inject(spi);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
+
if (useSsl) {
IgniteMock ignite = GridTestUtils.getFieldValue(spi, IgniteSpiAdapter.class, "ignite");
@@ -770,6 +801,14 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
* @throws Exception If failed.
*/
private void stopSpis() throws Exception {
+ if (timeoutProcessor != null) {
+ timeoutProcessor.onKernalStop(true);
+
+ timeoutProcessor.stop(true);
+
+ timeoutProcessor = null;
+ }
+
for (CommunicationSpi<Message> spi : spis) {
spi.onContextDestroyed();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
index 3f58055..7b59da3 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
@@ -29,6 +29,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.nio.GridNioRecoveryDescriptor;
import org.apache.ignite.internal.util.nio.GridNioServer;
@@ -47,6 +48,7 @@ import org.apache.ignite.spi.communication.GridTestMessage;
import org.apache.ignite.testframework.GridSpiTestContext;
import org.apache.ignite.testframework.GridTestNode;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.IgniteTestResources;
import org.apache.ignite.testframework.junits.spi.GridSpiAbstractTest;
import org.apache.ignite.testframework.junits.spi.GridSpiTest;
@@ -70,6 +72,9 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
/** */
private static final int SPI_CNT = 2;
+ /** */
+ private static GridTimeoutProcessor timeoutProcessor;
+
/**
*
*/
@@ -98,8 +103,6 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
/** {@inheritDoc} */
@Override public void onMessage(UUID nodeId, Message msg, IgniteRunnable msgC) {
- info("Test listener received message: " + msg);
-
assertTrue("Unexpected message: " + msg, msg instanceof GridTestMessage);
GridTestMessage msg0 = (GridTestMessage)msg;
@@ -171,6 +174,17 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
spi0.sendMessage(node1, new GridTestMessage(node0.id(), ++msgId, 0), ackC);
spi1.sendMessage(node0, new GridTestMessage(node1.id(), ++msgId, 0), ackC);
+
+ if (j == 0) {
+ final TestListener lsnr0 = (TestListener)spi0.getListener();
+ final TestListener lsnr1 = (TestListener)spi1.getListener();
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return lsnr0.rcvCnt.get() >= 1 && lsnr1.rcvCnt.get() >= 1;
+ }
+ }, 1000);
+ }
}
expMsgs += msgPerIter;
@@ -415,6 +429,12 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
Map<ClusterNode, GridSpiTestContext> ctxs = new HashMap<>();
+ timeoutProcessor = new GridTimeoutProcessor(new GridTestKernalContext(log));
+
+ timeoutProcessor.start();
+
+ timeoutProcessor.onKernalStart();
+
for (int i = 0; i < SPI_CNT; i++) {
TcpCommunicationSpi spi = getSpi(ackCnt, idleTimeout, queueLimit);
@@ -428,6 +448,8 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
ctx.setLocalNode(node);
+ ctx.timeoutProcessor(timeoutProcessor);
+
spiRsrcs.add(rsrcs);
rsrcs.inject(spi);
@@ -436,6 +458,8 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
node.setAttributes(spi.getNodeAttributes());
+ node.order(i);
+
nodes.add(node);
spi.spiStart(getTestIgniteInstanceName() + (i + 1));
@@ -491,6 +515,14 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
* @throws Exception If failed.
*/
private void stopSpis() throws Exception {
+ if (timeoutProcessor != null) {
+ timeoutProcessor.onKernalStop(true);
+
+ timeoutProcessor.stop(true);
+
+ timeoutProcessor = null;
+ }
+
for (CommunicationSpi<Message> spi : spis) {
spi.onContextDestroyed();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiDropNodesTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiDropNodesTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiDropNodesTest.java
index 2b49d53..9c59cb2 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiDropNodesTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiDropNodesTest.java
@@ -188,8 +188,7 @@ public class TcpCommunicationSpiDropNodesTest extends GridCommonAbstractTest {
final CountDownLatch latch = new CountDownLatch(1);
grid(0).events().localListen(new IgnitePredicate<Event>() {
- @Override
- public boolean apply(Event event) {
+ @Override public boolean apply(Event evt) {
latch.countDown();
return true;
@@ -239,14 +238,14 @@ public class TcpCommunicationSpiDropNodesTest extends GridCommonAbstractTest {
}, 5000);
try {
- fut1.get();
+ fut1.get(1000);
}
catch (IgniteCheckedException e) {
// No-op.
}
try {
- fut2.get();
+ fut2.get(1000);
}
catch (IgniteCheckedException e) {
// No-op.
@@ -297,8 +296,9 @@ public class TcpCommunicationSpiDropNodesTest extends GridCommonAbstractTest {
*/
private static class TestCommunicationSpi extends TcpCommunicationSpi {
/** {@inheritDoc} */
- @Override protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx)
- throws IgniteCheckedException {
+ @Override protected IgniteInternalFuture<GridCommunicationClient> createTcpClient(ClusterNode node, int connIdx)
+ throws IgniteCheckedException
+ {
if (pred.apply(getLocalNode(), node)) {
Map<String, Object> attrs = new HashMap<>(node.attributes());
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiFaultyClientTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiFaultyClientTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiFaultyClientTest.java
index 4fe67c1..baa1270 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiFaultyClientTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiFaultyClientTest.java
@@ -240,8 +240,9 @@ public class TcpCommunicationSpiFaultyClientTest extends GridCommonAbstractTest
*/
private static class TestCommunicationSpi extends TcpCommunicationSpi {
/** {@inheritDoc} */
- @Override protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx)
- throws IgniteCheckedException {
+ @Override protected IgniteInternalFuture<GridCommunicationClient> createTcpClient(ClusterNode node, int connIdx)
+ throws IgniteCheckedException
+ {
if (PRED.apply(node)) {
Map<String, Object> attrs = new HashMap<>(node.attributes());
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopExternalCommunication.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopExternalCommunication.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopExternalCommunication.java
index 8a20eec..a241a04 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopExternalCommunication.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopExternalCommunication.java
@@ -1004,7 +1004,10 @@ public class HadoopExternalCommunication {
HandshakeFinish fin = new HandshakeFinish();
- GridNioSession ses = nioSrvr.createSession(ch, F.asMap(HANDSHAKE_FINISH_META, fin)).get();
+ GridNioFuture<GridNioSession> sesFut =
+ nioSrvr.createSession(ch, F.<Integer, Object>asMap(HANDSHAKE_FINISH_META, fin), false, null);
+
+ GridNioSession ses = sesFut.get();
client = new HadoopTcpNioCommunicationClient(ses);
[8/9] ignite git commit: ignite-4003 Async outgoing connections for
communication SPI
Posted by ag...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
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 42879b7..9622c84 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
@@ -35,6 +35,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -46,8 +47,6 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
@@ -66,7 +65,7 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.util.GridConcurrentFactory;
-import org.apache.ignite.internal.util.GridSpinReadWriteLock;
+import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.ipc.IpcEndpoint;
import org.apache.ignite.internal.util.ipc.IpcToNioAdapter;
@@ -78,6 +77,7 @@ import org.apache.ignite.internal.util.nio.GridConnectionBytesVerifyFilter;
import org.apache.ignite.internal.util.nio.GridDirectParser;
import org.apache.ignite.internal.util.nio.GridNioCodecFilter;
import org.apache.ignite.internal.util.nio.GridNioFilter;
+import org.apache.ignite.internal.util.nio.GridNioFuture;
import org.apache.ignite.internal.util.nio.GridNioMessageReaderFactory;
import org.apache.ignite.internal.util.nio.GridNioMessageTracker;
import org.apache.ignite.internal.util.nio.GridNioMessageWriterFactory;
@@ -90,9 +90,7 @@ import org.apache.ignite.internal.util.nio.GridNioSession;
import org.apache.ignite.internal.util.nio.GridNioSessionMetaKey;
import org.apache.ignite.internal.util.nio.GridShmemCommunicationClient;
import org.apache.ignite.internal.util.nio.GridTcpNioCommunicationClient;
-import org.apache.ignite.internal.util.nio.ssl.BlockingSslHandler;
import org.apache.ignite.internal.util.nio.ssl.GridNioSslFilter;
-import org.apache.ignite.internal.util.nio.ssl.GridSslMeta;
import org.apache.ignite.internal.util.typedef.CI2;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
@@ -106,7 +104,6 @@ import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -136,7 +133,6 @@ import org.jsr166.LongAdder8;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
-import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.SSL_META;
/**
* <tt>TcpCommunicationSpi</tt> is default communication SPI which uses
@@ -297,10 +293,15 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
/** Connection index meta for session. */
private static final int CONN_IDX_META = GridNioSessionMetaKey.nextUniqueKey();
+ /** Recovery descriptor meta key. */
+ private static final int RECOVERY_DESC_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
/** Message tracker meta for session. */
private static final int TRACKER_META = GridNioSessionMetaKey.nextUniqueKey();
+ /** Connection context meta key. */
+ private static final int CONN_CTX_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
+
/**
* Default local port range (value is <tt>100</tt>).
* See {@link #setLocalPortRange(int)} for details.
@@ -335,6 +336,12 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
/** Handshake message type. */
public static final short HANDSHAKE_MSG_TYPE = -3;
+ /** Ignite header message. */
+ private static final Message IGNITE_HEADER_MSG = new IgniteHeaderMessage();
+
+ /** Skip ack. For test purposes only. */
+ private boolean skipAck;
+
/** */
private ConnectGateway connectGate;
@@ -408,10 +415,10 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
log.debug("Session was closed but there are unacknowledged messages, " +
"will try to reconnect [rmtNode=" + outDesc.node().id() + ']');
- DisconnectedSessionInfo disconnectData =
- new DisconnectedSessionInfo(outDesc, connId.connectionIndex());
+ SessionInfo sesInfo =
+ new SessionInfo(ses, connId.connectionIndex(), SessionState.RECONNECT);
- commWorker.addProcessDisconnectRequest(disconnectData);
+ commWorker.addSessionStateChangeRequest(sesInfo);
}
}
else
@@ -437,14 +444,69 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
if (msg instanceof NodeIdMessage) {
sndId = U.bytesToUuid(((NodeIdMessage) msg).nodeIdBytes, 0);
- connKey = new ConnectionKey(sndId, 0, -1);
+
+ if (ses.remoteAddress() != null) { // Not shmem.
+ assert !ses.accepted();
+
+ ConnectContext ctx = ses.meta(CONN_CTX_META_KEY);
+
+ assert ctx != null;
+ assert ctx.expNodeId != null;
+
+ if (sndId.equals(ctx.expNodeId)) {
+ GridNioRecoveryDescriptor recoveryDesc = ses.outRecoveryDescriptor();
+
+ assert recoveryDesc != null;
+
+ long connCnt = recoveryDesc.incrementConnectCount();
+
+ connKey = new ConnectionKey(sndId, ctx.connIdx, connCnt);
+
+ final ConnectionKey old = ses.addMeta(CONN_IDX_META, connKey);
+
+ assert old == null;
+
+ ses.send(IGNITE_HEADER_MSG);
+
+ ClusterNode locNode = getLocalNode();
+
+ if (locNode == null) {
+ commWorker.addSessionStateChangeRequest(new SessionInfo(ses, SessionState.CLOSE,
+ new IgniteCheckedException("Local node has not been started or " +
+ "fully initialized [isStopping=" + getSpiContext().isStopping() + ']')));
+
+ return;
+ }
+
+ int handshakeConnIdx = connPlc.connectionIndex();
+
+ HandshakeMessage handshakeMsg = new HandshakeMessage2(locNode.id(), connCnt,
+ recoveryDesc.received(), handshakeConnIdx);
+
+ if (log.isDebugEnabled())
+ log.debug("Write handshake message [rmtNode=" + sndId +
+ ", msg=" + handshakeMsg + ']');
+
+ ses.send(handshakeMsg);
+ }
+ else {
+ commWorker.addSessionStateChangeRequest(new SessionInfo(ses, SessionState.CLOSE,
+ new IgniteCheckedException("Remote node ID is not as expected [expected=" +
+ ctx.expNodeId + ", rcvd=" + sndId + ']')));
+ }
+
+ return;
+ }
+ else
+ connKey = new ConnectionKey(sndId, 0, -1);
}
else {
assert msg instanceof HandshakeMessage : msg;
HandshakeMessage msg0 = (HandshakeMessage)msg;
- sndId = ((HandshakeMessage)msg).nodeId();
+ sndId = msg0.nodeId();
+
connKey = new ConnectionKey(sndId, msg0.connectionIndex(), msg0.connectCount());
}
@@ -485,30 +547,21 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
if (reserve)
connectedNew(recoveryDesc, ses, true);
else {
- if (c.failed) {
- ses.send(new RecoveryLastReceivedMessage(-1));
-
- for (GridNioSession ses0 : nioSrvr.sessions()) {
- ConnectionKey key0 = ses0.meta(CONN_IDX_META);
-
- if (ses0.accepted() && key0 != null &&
- key0.nodeId().equals(connKey.nodeId()) &&
- key0.connectionIndex() == connKey.connectionIndex() &&
- key0.connectCount() < connKey.connectCount())
- ses0.close();
- }
+ for (GridNioSession ses0 : nioSrvr.sessions()) {
+ ConnectionKey key0 = ses0.meta(CONN_IDX_META);
+
+ if (ses0.accepted() && key0 != null &&
+ key0.nodeId().equals(connKey.nodeId()) &&
+ key0.connectionIndex() == connKey.connectionIndex() &&
+ key0.connectCount() < connKey.connectCount())
+ ses0.close();
}
}
}
else {
assert connKey.connectionIndex() >= 0 : connKey;
- GridCommunicationClient[] curClients = clients.get(sndId);
-
- GridCommunicationClient oldClient =
- curClients != null && connKey.connectionIndex() < curClients.length ?
- curClients[connKey.connectionIndex()] :
- null;
+ GridCommunicationClient oldClient = nodeClient(sndId, connKey.connectionIndex());
boolean hasShmemClient = false;
@@ -537,10 +590,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
final GridNioRecoveryDescriptor recoveryDesc = inRecoveryDescriptor(rmtNode, connKey);
if (oldFut == null) {
- curClients = clients.get(sndId);
-
- oldClient = curClients != null && connKey.connectionIndex() < curClients.length ?
- curClients[connKey.connectionIndex()] : null;
+ oldClient = nodeClient(sndId, connKey.connectionIndex());
if (oldClient != null) {
if (oldClient instanceof GridTcpNioCommunicationClient) {
@@ -584,7 +634,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
}
}
else {
- if (oldFut instanceof ConnectFuture && locNode.order() < rmtNode.order()) {
+ if (oldFut instanceof ReserveClientFuture && locNode.order() < rmtNode.order()) {
if (log.isDebugEnabled()) {
log.debug("Received incoming connection from remote node while " +
"connecting to this node, rejecting [locNode=" + locNode.id() +
@@ -606,17 +656,31 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
}
}
- @Override public void onMessage(GridNioSession ses, Message msg) {
+ @Override public void onMessage(final GridNioSession ses, Message msg) {
ConnectionKey connKey = ses.meta(CONN_IDX_META);
if (connKey == null) {
- assert ses.accepted() : ses;
-
- if (!connectGate.tryEnter()) {
+ if (ses.accepted() && !connectGate.tryEnter()) { // Outgoing connection already entered gate.
if (log.isDebugEnabled())
log.debug("Close incoming connection, failed to enter gateway.");
- ses.close();
+ try {
+ nioSrvr.sendSystem(ses, new RecoveryLastReceivedMessage(-1), new IgniteInClosure<IgniteInternalFuture<?>>() {
+ @Override public void apply(IgniteInternalFuture<?> fut) {
+ try {
+ fut.get();
+
+ ses.close();
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send last received message: " + e, e);
+ }
+ }
+ });
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send message: " + e, e);
+ }
return;
}
@@ -625,7 +689,8 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
onFirstMessage(ses, msg);
}
finally {
- connectGate.leave();
+ if (ses.accepted())
+ connectGate.leave();
}
}
else {
@@ -637,13 +702,49 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
if (recovery != null) {
RecoveryLastReceivedMessage msg0 = (RecoveryLastReceivedMessage)msg;
+ long rcvCnt = msg0.received();
+
if (log.isDebugEnabled()) {
log.debug("Received recovery acknowledgement [rmtNode=" + connKey.nodeId() +
", connIdx=" + connKey.connectionIndex() +
", rcvCnt=" + msg0.received() + ']');
}
- recovery.ackReceived(msg0.received());
+ ConnectContext ctx = ses.meta(CONN_CTX_META_KEY);
+
+ if (!ses.accepted() && ctx != null && ctx.rcvCnt == Long.MIN_VALUE) {
+ HandshakeTimeoutObject timeoutObj = ctx.handshakeTimeoutObj;
+
+ Exception err = null;
+
+ if (timeoutObj != null && !cancelHandshakeTimeout(timeoutObj)) {
+ err = new HandshakeTimeoutException("Failed to perform handshake due to timeout " +
+ "(consider increasing 'connectionTimeout' configuration property).");
+ }
+
+ if (rcvCnt == -1 || err != null) {
+ if (ses.remoteAddress() != null) {
+ SessionInfo sesInfo = new SessionInfo(ses, SessionState.CLOSE, err);
+
+ commWorker.addSessionStateChangeRequest(sesInfo);
+ }
+ }
+ else {
+ ctx.rcvCnt = rcvCnt;
+
+ recovery.onHandshake(rcvCnt);
+
+ nioSrvr.resend(ses);
+
+ recovery.onConnected();
+
+ SessionInfo sesInfo = new SessionInfo(ses, connKey.idx, SessionState.READY);
+
+ commWorker.addSessionStateChangeRequest(sesInfo);
+ }
+ }
+ else
+ recovery.ackReceived(rcvCnt);
return;
}
@@ -661,7 +762,8 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
", rcvCnt=" + rcvCnt + ']');
}
- ses.systemMessage(new RecoveryLastReceivedMessage(rcvCnt));
+ if (!skipAck)
+ ses.systemMessage(new RecoveryLastReceivedMessage(rcvCnt));
recovery.lastAcknowledged(rcvCnt);
}
@@ -712,7 +814,8 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
assert connKey != null && connKey.connectionIndex() >= 0 : connKey;
assert !usePairedConnections(node);
- recovery.onHandshake(rcvCnt);
+ if (ses.accepted())
+ recovery.onHandshake(rcvCnt);
ses.inRecoveryDescriptor(recovery);
ses.outRecoveryDescriptor(recovery);
@@ -778,9 +881,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
/** */
private final ClusterNode rmtNode;
- /** */
- private boolean failed;
-
/**
* @param ses Incoming session.
* @param recoveryDesc Recovery descriptor.
@@ -797,8 +897,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
/** {@inheritDoc} */
@Override public void apply(Boolean success) {
try {
- failed = !success;
-
if (success) {
IgniteInClosure<IgniteInternalFuture<?>> lsnr = new IgniteInClosure<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> msgFut) {
@@ -919,10 +1017,28 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
}
else {
try {
- fut.onDone();
+ nioSrvr.sendSystem(ses, new RecoveryLastReceivedMessage(-1), new IgniteInClosure<IgniteInternalFuture<?>>() {
+ @Override public void apply(IgniteInternalFuture<?> msgFut) {
+ try {
+ msgFut.get();
+ } catch (IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to send recovery handshake " +
+ "[rmtNode=" + rmtNode.id() + ", err=" + e + ']');
+
+ recoveryDesc.release();
+ } finally {
+ fut.onDone();
+
+ clientFuts.remove(connKey, fut);
+
+ ses.close();
+ }
+ }
+ });
}
- finally {
- clientFuts.remove(connKey, fut);
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send message: " + e, e);
}
}
}
@@ -1727,12 +1843,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
nioSrvr.dumpStats();
}
- /** */
- private final ThreadLocal<Integer> threadConnIdx = new ThreadLocal<>();
-
- /** */
- private final AtomicInteger connIdx = new AtomicInteger();
-
/** {@inheritDoc} */
@Override public Map<String, Object> getNodeAttributes() throws IgniteSpiException {
initFailureDetectionTimeout();
@@ -1988,9 +2098,20 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
assert formatter != null;
+ UUID rmtNodeId = null;
+
ConnectionKey key = ses.meta(CONN_IDX_META);
- return key != null ? formatter.writer(key.nodeId()) : null;
+ if (key != null)
+ rmtNodeId = key.nodeId();
+ else {
+ ConnectContext ctx = ses.meta(CONN_CTX_META_KEY);
+
+ if (ctx != null)
+ rmtNodeId = ctx.expNodeId;
+ }
+
+ return key != null ? formatter.writer(rmtNodeId) : null;
}
};
@@ -2000,7 +2121,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
IgnitePredicate<Message> skipRecoveryPred = new IgnitePredicate<Message>() {
@Override public boolean apply(Message msg) {
- return msg instanceof RecoveryLastReceivedMessage;
+ return msg instanceof NotRecoverable;
}
};
@@ -2339,48 +2460,108 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
int connIdx = connPlc.connectionIndex();
+ send(node, connIdx, msg, ackC);
+ }
+ }
+
+ /**
+ * Try to send message.
+ */
+ private void send(final ClusterNode node,
+ final int connIdx,
+ final Message msg,
+ final IgniteInClosure<IgniteException> ackC
+ ) {
+ final GridCommunicationClient client = nodeClient(node.id(), connIdx);
+
+ if (client != null && client.reserve()) {
try {
- boolean retry;
+ send0(client, node, msg, ackC);
+ }
+ catch (IgniteCheckedException e) {
+ if (removeNodeClient(node.id(), client))
+ client.forceClose();
- do {
- client = reserveClient(node, connIdx);
+ throw new IgniteSpiException("Failed to send message to remote node: " + node, e);
+ }
+ }
+ else {
+ if (client != null)
+ removeNodeClient(node.id(), client);
- UUID nodeId = null;
+ IgniteInternalFuture<GridCommunicationClient> clientFut = reserveClient(node, connIdx);
- if (!client.async())
- nodeId = node.id();
+ clientFut.listen(new IgniteInClosure<IgniteInternalFuture<GridCommunicationClient>>() {
+ @Override public void apply(IgniteInternalFuture<GridCommunicationClient> fut) {
+ GridCommunicationClient client = null;
- retry = client.sendMessage(nodeId, msg, ackC);
+ try {
+ client = fut.get();
- client.release();
+ send0(client, node, msg, ackC);
+ }
+ catch (IgniteCheckedException e) {
+ LT.error(log, e, "Unexpected error occurred during sending of message to node: " + node.id());
- if (!retry)
- sentMsgsCnt.increment();
- else {
- removeNodeClient(node.id(), client);
+ if (client != null && removeNodeClient(node.id(), client))
+ client.forceClose();
+ }
+ }
+ });
+ }
+ }
- ClusterNode node0 = getSpiContext().node(node.id());
+ /**
+ * @param client Client.
+ * @param node Node.
+ * @param msg Message.
+ * @param ackC Ack closure.
+ */
+ private void send0(
+ GridCommunicationClient client,
+ ClusterNode node,
+ Message msg,
+ IgniteInClosure<IgniteException> ackC
+ ) throws IgniteCheckedException {
+ assert client != null;
- if (node0 == null)
- throw new IgniteCheckedException("Failed to send message to remote node " +
- "(node has left the grid): " + node.id());
- }
+ UUID nodeId = null;
- client = null;
- }
- while (retry);
- }
- catch (IgniteCheckedException e) {
- throw new IgniteSpiException("Failed to send message to remote node: " + node, e);
- }
- finally {
- if (client != null && removeNodeClient(node.id(), client))
- client.forceClose();
+ if (!client.async())
+ nodeId = node.id();
+
+ boolean retry = client.sendMessage(nodeId, msg, ackC);
+
+ client.release();
+
+ if (!retry)
+ sentMsgsCnt.increment();
+ else {
+ removeNodeClient(node.id(), client);
+
+ ClusterNode node0 = getSpiContext().node(node.id());
+
+ if (node0 == null) {
+ U.warn(log, "Failed to send message to remote node (node has left the grid): " + node.id());
+
+ return;
}
+
+ send(node, client.connectionIndex(), msg, ackC);
}
}
/**
+ * @param nodeId Node id.
+ * @param connIdx Connection index.
+ */
+ private GridCommunicationClient nodeClient(UUID nodeId, int connIdx) {
+ GridCommunicationClient[] curClients = clients.get(nodeId);
+
+ return curClients != null && connIdx < curClients.length ? curClients[connIdx] : null;
+ }
+
+ /**
* @param nodeId Node ID.
* @param rmvClient Client to remove.
* @return {@code True} if client was removed.
@@ -2449,95 +2630,245 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
* @param node Node to which client should be open.
* @param connIdx Connection index.
* @return The existing or just created client.
- * @throws IgniteCheckedException Thrown if any exception occurs.
*/
- private GridCommunicationClient reserveClient(ClusterNode node, int connIdx) throws IgniteCheckedException {
- assert node != null;
- assert (connIdx >= 0 && connIdx < connectionsPerNode) || !usePairedConnections(node) : connIdx;
+ private IgniteInternalFuture<GridCommunicationClient> reserveClient(ClusterNode node, int connIdx) {
+ GridFutureAdapter<GridCommunicationClient> fut = new GridFutureAdapter<>();
- UUID nodeId = node.id();
+ tryReserveClient(node, connIdx, fut);
- while (true) {
- GridCommunicationClient[] curClients = clients.get(nodeId);
+ return fut;
+ }
+
+ /**
+ * @param node Node.
+ * @param connIdx Connection index.
+ * @param fut Future.
+ */
+ private void tryReserveClient(
+ final ClusterNode node,
+ final int connIdx,
+ final GridFutureAdapter<GridCommunicationClient> fut)
+ {
+ final ReserveClientFuture reserveFut = new ReserveClientFuture(node, connIdx);
+
+ reserveFut.listen(new IgniteInClosure<IgniteInternalFuture<GridCommunicationClient>>() {
+ @Override public void apply(IgniteInternalFuture<GridCommunicationClient> fut0) {
+ try {
+ GridCommunicationClient client = fut0.get();
+
+ if (client != null)
+ fut.onDone(client);
+ else
+ tryReserveClient(node, connIdx, fut);
+ }
+ catch (IgniteCheckedException e) {
+ fut.onDone(e);
+ }
+ }
+ });
+
+ try {
+ reserveFut.reserve();
+ }
+ catch (Exception e) {
+ fut.onDone(e);
+ }
+ }
+
+ /**
+ *
+ */
+ private class ReserveClientFuture extends GridFutureAdapter<GridCommunicationClient> {
+ /** Node. */
+ private final ClusterNode node;
+
+ /** Connection index. */
+ private final int connIdx;
+
+ /**
+ * @param node Node.
+ */
+ ReserveClientFuture(ClusterNode node, int connIdx) {
+ assert node != null;
+ assert (connIdx >= 0 && connIdx < connectionsPerNode) || !usePairedConnections(node) : connIdx;
+
+ this.node = node;
+ this.connIdx = connIdx;
+ }
+
+ /**
+ *
+ */
+ void reserve() {
+ final UUID nodeId = node.id();
- GridCommunicationClient client = curClients != null && connIdx < curClients.length ?
- curClients[connIdx] : null;
+ final GridCommunicationClient client = nodeClient(nodeId, connIdx);
+
+ final GridFutureAdapter<GridCommunicationClient> connFut;
if (client == null) {
- if (stopping)
- throw new IgniteSpiException("Node is stopping.");
+ if (stopping) {
+ onDone(new IgniteSpiException("Node is stopping."));
+
+ return;
+ }
// Do not allow concurrent connects.
- GridFutureAdapter<GridCommunicationClient> fut = new ConnectFuture();
+ connFut = this;
- ConnectionKey connKey = new ConnectionKey(nodeId, connIdx, -1);
+ final ConnectionKey connKey = new ConnectionKey(nodeId, connIdx, -1);
- GridFutureAdapter<GridCommunicationClient> oldFut = clientFuts.putIfAbsent(connKey, fut);
+ final GridFutureAdapter<GridCommunicationClient> oldFut = clientFuts.putIfAbsent(connKey, connFut);
if (oldFut == null) {
try {
- GridCommunicationClient[] curClients0 = clients.get(nodeId);
-
- GridCommunicationClient client0 = curClients0 != null && connIdx < curClients0.length ?
- curClients0[connIdx] : null;
+ GridCommunicationClient client0 = nodeClient(nodeId, connIdx);
if (client0 == null) {
- client0 = createNioClient(node, connIdx);
-
- if (client0 != null) {
- addNodeClient(node, connIdx, client0);
+ IgniteInternalFuture<GridCommunicationClient> clientFut = createNioClient(node, connIdx);
- if (client0 instanceof GridTcpNioCommunicationClient) {
- GridTcpNioCommunicationClient tcpClient = ((GridTcpNioCommunicationClient)client0);
-
- if (tcpClient.session().closeTime() > 0 && removeNodeClient(nodeId, client0)) {
- if (log.isDebugEnabled())
- log.debug("Session was closed after client creation, will retry " +
- "[node=" + node + ", client=" + client0 + ']');
+ clientFut.listen(new IgniteInClosure<IgniteInternalFuture<GridCommunicationClient>>() {
+ @Override public void apply(IgniteInternalFuture<GridCommunicationClient> fut) {
+ try {
+ GridCommunicationClient client0 = fut.get();
+
+ if (client0 != null) {
+ addNodeClient(node, connIdx, client0);
+
+ if (client0 instanceof GridTcpNioCommunicationClient) {
+ GridTcpNioCommunicationClient tcpClient =
+ ((GridTcpNioCommunicationClient)client0);
+
+ if (tcpClient.session().closeTime() > 0 && removeNodeClient(nodeId, client0)) {
+ if (log.isDebugEnabled())
+ log.debug("Session was closed after client creation, " +
+ "will retry [node=" + node + ", client=" + client0 + ']');
+
+ client0 = null;
+ }
+ }
+
+ if (client0 == null) {
+ clientFuts.remove(connKey, connFut);
+
+ onDone();
+ }
+ else if (client0.reserve()) {
+ clientFuts.remove(connKey, connFut);
+
+ onDone(client0);
+ }
+ else {
+ clientFuts.remove(connKey, connFut);
+
+ removeNodeClient(nodeId, client0);
+
+ onDone();
+ }
+ }
+ else {
+ final long currTime = U.currentTimeMillis();
+
+ addTimeoutObject(new IgniteSpiTimeoutObject() {
+ private final IgniteUuid id = IgniteUuid.randomUuid();
+
+ @Override public IgniteUuid id() {
+ return id;
+ }
+
+ @Override public long endTime() {
+ return currTime + 200;
+ }
+
+ @Override public void onTimeout() {
+ SessionInfo sesInfo = new SessionInfo(null, SessionState.RETRY,
+ new Runnable() {
+ @Override public void run() {
+ clientFuts.remove(connKey, connFut);
+
+ onDone();
+ }
+ });
+
+ commWorker.addSessionStateChangeRequest(sesInfo);
+ }
+ });
+ }
+ }
+ catch (IgniteCheckedException e) {
+ clientFuts.remove(connKey, connFut);
- client0 = null;
+ onDone(e);
}
}
- }
- else
- U.sleep(200);
+ });
}
+ else {
+ assert connIdx == client0.connectionIndex() : client0;
- fut.onDone(client0);
+ if (client0.reserve())
+ onDone(client0);
+ else {
+ removeNodeClient(nodeId, client0);
+
+ onDone();
+ }
+ }
}
catch (Throwable e) {
- fut.onDone(e);
+ connFut.onDone(e);
if (e instanceof Error)
throw (Error)e;
}
- finally {
- clientFuts.remove(connKey, fut);
- }
}
- else
- fut = oldFut;
+ else {
+ oldFut.listen(new IgniteInClosure<IgniteInternalFuture<GridCommunicationClient>>() {
+ @Override public void apply(IgniteInternalFuture<GridCommunicationClient> fut) {
+ try {
+ GridCommunicationClient client0 = fut.get();
- client = fut.get();
+ if (client0 == null) {
+ clientFuts.remove(connKey, oldFut);
- if (client == null)
- continue;
+ onDone();
+ }
+ else if (client0.reserve()) {
+ clientFuts.remove(connKey, oldFut);
- if (getSpiContext().node(nodeId) == null) {
- if (removeNodeClient(nodeId, client))
- client.forceClose();
+ onDone(client0);
+ }
+ else {
+ clientFuts.remove(connKey, oldFut);
+
+ removeNodeClient(nodeId, client0);
- throw new IgniteSpiException("Destination node is not in topology: " + node.id());
+ onDone();
+ }
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
+ }
+ });
}
}
+ else {
+ assert connIdx == client.connectionIndex() : client;
+
+ if (client.reserve())
+ onDone(client);
+ else {
+ removeNodeClient(nodeId, client);
- assert connIdx == client.connectionIndex() : client;
+ onDone();
+ }
+ }
+ }
- if (client.reserve())
- return client;
- else
- // Client has just been closed by idle worker. Help it and try again.
- removeNodeClient(nodeId, client);
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(ReserveClientFuture.class, this);
}
}
@@ -2545,10 +2876,8 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
* @param node Node to create client for.
* @param connIdx Connection index.
* @return Client.
- * @throws IgniteCheckedException If failed.
*/
- @Nullable private GridCommunicationClient createNioClient(ClusterNode node, int connIdx)
- throws IgniteCheckedException {
+ protected IgniteInternalFuture<GridCommunicationClient> createNioClient(ClusterNode node, int connIdx) {
assert node != null;
Integer shmemPort = node.attribute(createSpiAttributeName(ATTR_SHMEM_PORT));
@@ -2556,7 +2885,9 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
ClusterNode locNode = getSpiContext().localNode();
if (locNode == null)
- throw new IgniteCheckedException("Failed to create NIO client (local node is stopping)");
+ return new GridFinishedFuture<>(
+ new IgniteCheckedException("Failed to create NIO client (local node is stopping)")
+ );
if (log.isDebugEnabled())
log.debug("Creating NIO client to node: " + node);
@@ -2573,7 +2904,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
if (log.isDebugEnabled())
log.debug("Shmem client created: " + client);
- return client;
+ return new GridFinishedFuture<>(client);
}
catch (IgniteCheckedException e) {
if (e.hasCause(IpcOutOfSystemResourcesException.class))
@@ -2584,21 +2915,17 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
else if (log.isDebugEnabled())
log.debug("Failed to establish shared memory connection with local node (node has left): " +
node.id());
+
+ return new GridFinishedFuture<>(e);
}
}
- connectGate.enter();
try {
- GridCommunicationClient client = createTcpClient(node, connIdx);
-
- if (log.isDebugEnabled())
- log.debug("TCP client created: " + client);
-
- return client;
+ return createTcpClient(node, connIdx);
}
- finally {
- connectGate.leave();
+ catch (IgniteCheckedException e) {
+ return new GridFinishedFuture<>(e);
}
}
@@ -2647,12 +2974,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
}
try {
- safeHandshake(client,
- null,
- node.id(),
- timeoutHelper.nextTimeoutChunk(connTimeout0),
- null,
- null);
+ safeHandshake(client, node.id(), timeoutHelper.nextTimeoutChunk(connTimeout0));
}
catch (HandshakeTimeoutException | IgniteSpiOperationTimeoutException e) {
client.forceClose();
@@ -2714,7 +3036,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
ConnectionKey id = ses.meta(CONN_IDX_META);
if (id != null) {
- ClusterNode node = getSpiContext().node(id.nodeId);
+ ClusterNode node = getSpiContext().node(id.nodeId());
if (node != null && node.isClient()) {
String msg = "Client node outbound message queue size exceeded slowClientQueueLimit, " +
@@ -2737,532 +3059,527 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
*
* @param node Remote node.
* @param connIdx Connection index.
- * @return Client.
+ * @return Client future.
* @throws IgniteCheckedException If failed.
*/
- protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx) throws IgniteCheckedException {
- Collection<String> rmtAddrs0 = node.attribute(createSpiAttributeName(ATTR_ADDRS));
- Collection<String> rmtHostNames0 = node.attribute(createSpiAttributeName(ATTR_HOST_NAMES));
- Integer boundPort = node.attribute(createSpiAttributeName(ATTR_PORT));
- Collection<InetSocketAddress> extAddrs = node.attribute(createSpiAttributeName(ATTR_EXT_ADDRS));
-
- boolean isRmtAddrsExist = (!F.isEmpty(rmtAddrs0) && boundPort != null);
- boolean isExtAddrsExist = !F.isEmpty(extAddrs);
+ protected IgniteInternalFuture<GridCommunicationClient> createTcpClient(ClusterNode node, int connIdx)
+ throws IgniteCheckedException
+ {
+ TcpClientFuture fut = new TcpClientFuture(node, connIdx);
- if (!isRmtAddrsExist && !isExtAddrsExist)
- throw new IgniteCheckedException("Failed to send message to the destination node. Node doesn't have any " +
- "TCP communication addresses or mapped external addresses. Check configuration and make sure " +
- "that you use the same communication SPI on all nodes. Remote node id: " + node.id());
+ connectGate.enter();
- LinkedHashSet<InetSocketAddress> addrs;
+ fut.connect();
- // Try to connect first on bound addresses.
- if (isRmtAddrsExist) {
- List<InetSocketAddress> addrs0 = new ArrayList<>(U.toSocketAddresses(rmtAddrs0, rmtHostNames0, boundPort));
+ fut.listen(new IgniteInClosure<IgniteInternalFuture<GridCommunicationClient>>() {
+ @Override public void apply(IgniteInternalFuture<GridCommunicationClient> fut0) {
+ connectGate.leave();
+ }
+ });
- boolean sameHost = U.sameMacs(getSpiContext().localNode(), node);
+ return fut;
+ }
- Collections.sort(addrs0, U.inetAddressesComparator(sameHost));
+ /**
+ * @param timeoutObj Timeout object.
+ */
+ private boolean cancelHandshakeTimeout(HandshakeTimeoutObject timeoutObj) {
+ boolean cancelled = timeoutObj.cancel();
- addrs = new LinkedHashSet<>(addrs0);
- }
- else
- addrs = new LinkedHashSet<>();
+ if (cancelled)
+ removeTimeoutObject(timeoutObj);
- // Then on mapped external addresses.
- if (isExtAddrsExist)
- addrs.addAll(extAddrs);
+ return cancelled;
+ }
- Set<InetAddress> allInetAddrs = U.newHashSet(addrs.size());
+ /**
+ *
+ */
+ private class TcpClientFuture extends GridFutureAdapter<GridCommunicationClient> {
+ /** Node. */
+ private final ClusterNode node;
- for (InetSocketAddress addr : addrs)
- allInetAddrs.add(addr.getAddress());
+ /** Timeout helper. */
+ private final IgniteSpiOperationTimeoutHelper timeoutHelper =
+ new IgniteSpiOperationTimeoutHelper(TcpCommunicationSpi.this);
- List<InetAddress> reachableInetAddrs = U.filterReachable(allInetAddrs);
+ /** Addresses. */
+ private Collection<InetSocketAddress> addrs;
- if (reachableInetAddrs.size() < allInetAddrs.size()) {
- LinkedHashSet<InetSocketAddress> addrs0 = U.newLinkedHashSet(addrs.size());
+ /** Addresses it. */
+ private Iterator<InetSocketAddress> addrsIt;
- for (InetSocketAddress addr : addrs) {
- if (reachableInetAddrs.contains(addr.getAddress()))
- addrs0.add(addr);
- }
- for (InetSocketAddress addr : addrs) {
- if (!reachableInetAddrs.contains(addr.getAddress()))
- addrs0.add(addr);
- }
+ /** Current addresses. */
+ private volatile InetSocketAddress currAddr;
- addrs = addrs0;
- }
+ /** Err. */
+ private volatile IgniteCheckedException err;
- if (log.isDebugEnabled())
- log.debug("Addresses to connect for node [rmtNode=" + node.id() + ", addrs=" + addrs.toString() + ']');
+ /** Connect attempts. */
+ private volatile int connectAttempts;
- boolean conn = false;
- GridCommunicationClient client = null;
- IgniteCheckedException errs = null;
+ /** Attempts. */
+ private volatile int attempt;
- int connectAttempts = 1;
+ /** Connection index. */
+ private volatile int connIdx;
- for (InetSocketAddress addr : addrs) {
- long connTimeout0 = connTimeout;
+ /**
+ * @param node Node.
+ */
+ TcpClientFuture(ClusterNode node, int connIdx) {
+ this.node = node;
+ this.connIdx = connIdx;
+ }
- int attempt = 1;
+ /**
+ * Connects to remote node.
+ */
+ void connect() {
+ try {
+ addrs = addrs();
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
- IgniteSpiOperationTimeoutHelper timeoutHelper = new IgniteSpiOperationTimeoutHelper(this);
+ return;
+ }
- while (!conn) { // Reconnection on handshake timeout.
- try {
- SocketChannel ch = SocketChannel.open();
+ addrsIt = addrs.iterator();
- ch.configureBlocking(true);
+ tryConnect(true);
+ }
- ch.socket().setTcpNoDelay(tcpNoDelay);
- ch.socket().setKeepAlive(true);
+ /**
+ *
+ */
+ private void tryConnect(boolean next) {
+ if (next && !addrsIt.hasNext()) {
+ IgniteCheckedException err0 = err;
- if (sockRcvBuf > 0)
- ch.socket().setReceiveBufferSize(sockRcvBuf);
+ assert err0 != null;
- if (sockSndBuf > 0)
- ch.socket().setSendBufferSize(sockSndBuf);
+ UUID nodeId = node.id();
- if (getSpiContext().node(node.id()) == null) {
- U.closeQuiet(ch);
+ if (getSpiContext().node(nodeId) != null && (CU.clientNode(node) || !CU.clientNode(getLocalNode())) &&
+ X.hasCause(err, ConnectException.class, SocketTimeoutException.class, HandshakeTimeoutException.class,
+ IgniteSpiOperationTimeoutException.class))
+ {
+ LT.warn(log, "TcpCommunicationSpi failed to establish connection to node, node will be " +
+ "dropped from cluster [" + "rmtNode=" + node + ", err=" + err +
+ ", connectErrs=" + Arrays.toString(err.getSuppressed()) + ']');
- throw new ClusterTopologyCheckedException("Failed to send message " +
- "(node left topology): " + node);
- }
+ getSpiContext().failNode(nodeId, "TcpCommunicationSpi failed to establish connection to node " +
+ "[rmtNode=" + node + ", errs=" + err + ", connectErrs=" + Arrays.toString(err.getSuppressed()) + ']');
+ }
- ConnectionKey connKey = new ConnectionKey(node.id(), connIdx, -1);
+ onDone(err0);
- GridNioRecoveryDescriptor recoveryDesc = outRecoveryDescriptor(node, connKey);
+ return;
+ }
- if (!recoveryDesc.reserve()) {
- U.closeQuiet(ch);
+ if (next) {
+ attempt = 0;
- return null;
- }
+ connectAttempts = 0;
- long rcvCnt = -1;
+ currAddr = addrsIt.next();
+ }
- Map<Integer, Object> meta = new HashMap<>();
+ InetSocketAddress addr = currAddr;
- GridSslMeta sslMeta = null;
+ try {
+ final SocketChannel ch = SocketChannel.open();
- try {
- ch.socket().connect(addr, (int)timeoutHelper.nextTimeoutChunk(connTimeout));
+ ch.configureBlocking(false);
- if (isSslEnabled()) {
- meta.put(SSL_META.ordinal(), sslMeta = new GridSslMeta());
+ ch.socket().setTcpNoDelay(tcpNoDelay);
+ ch.socket().setKeepAlive(true);
- SSLEngine sslEngine = ignite.configuration().getSslContextFactory().create().createSSLEngine();
+ if (sockRcvBuf > 0)
+ ch.socket().setReceiveBufferSize(sockRcvBuf);
- sslEngine.setUseClientMode(true);
+ if (sockSndBuf > 0)
+ ch.socket().setSendBufferSize(sockSndBuf);
- sslMeta.sslEngine(sslEngine);
- }
+ if (getSpiContext().node(node.id()) == null) {
+ U.closeQuiet(ch);
- Integer handshakeConnIdx = connIdx;
+ onError(new ClusterTopologyCheckedException("Failed to send message " +
+ "(node left topology): " + node));
- rcvCnt = safeHandshake(ch,
- recoveryDesc,
- node.id(),
- timeoutHelper.nextTimeoutChunk(connTimeout0),
- sslMeta,
- handshakeConnIdx);
+ return;
+ }
- if (rcvCnt == -1)
- return null;
- }
- finally {
- if (recoveryDesc != null && rcvCnt == -1)
- recoveryDesc.release();
- }
+ final ConnectionKey connKey = new ConnectionKey(node.id(), connIdx, -1);
- try {
- meta.put(CONN_IDX_META, connKey);
+ final GridNioRecoveryDescriptor recoveryDesc = outRecoveryDescriptor(node, connKey);
- if (recoveryDesc != null) {
- recoveryDesc.onHandshake(rcvCnt);
+ if (!recoveryDesc.reserve()) {
+ U.closeQuiet(ch);
- meta.put(-1, recoveryDesc);
- }
+ onDone();
- GridNioSession ses = nioSrvr.createSession(ch, meta).get();
+ return;
+ }
- client = new GridTcpNioCommunicationClient(connIdx, ses, log);
+ final Map<Integer, Object> meta = new HashMap<>();
- conn = true;
- }
- finally {
- if (!conn) {
- if (recoveryDesc != null)
- recoveryDesc.release();
- }
- }
- }
- catch (HandshakeTimeoutException | IgniteSpiOperationTimeoutException e) {
- if (client != null) {
- client.forceClose();
+ final ConnectContext ctx = new ConnectContext();
- client = null;
- }
+ ctx.expNodeId = node.id();
- if (failureDetectionTimeoutEnabled() && (e instanceof HandshakeTimeoutException ||
- timeoutHelper.checkFailureTimeoutReached(e))) {
+ ctx.tcpClientFut = this;
- String msg = "Handshake timed out (failure detection timeout is reached) " +
- "[failureDetectionTimeout=" + failureDetectionTimeout() + ", addr=" + addr + ']';
+ ctx.connIdx = connIdx;
- onException(msg, e);
+ meta.put(CONN_CTX_META_KEY, ctx);
- if (log.isDebugEnabled())
- log.debug(msg);
+ meta.put(RECOVERY_DESC_META_KEY, recoveryDesc);
- if (errs == null)
- errs = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
- "Make sure that each ComputeTask and cache Transaction has a timeout set " +
- "in order to prevent parties from waiting forever in case of network issues " +
- "[nodeId=" + node.id() + ", addrs=" + addrs + ']');
+ final int timeoutChunk = (int)timeoutHelper.nextTimeoutChunk(connTimeout);
- errs.addSuppressed(new IgniteCheckedException("Failed to connect to address: " + addr, e));
+ final int attempt0 = attempt;
- break;
- }
+ final ConnectionTimeoutObject connTimeoutObj = new ConnectionTimeoutObject(ch, meta,
+ U.currentTimeMillis() + timeoutChunk * (1L << attempt0));
- assert !failureDetectionTimeoutEnabled();
+ addTimeoutObject(connTimeoutObj);
- onException("Handshake timed out (will retry with increased timeout) [timeout=" + connTimeout0 +
- ", addr=" + addr + ']', e);
+ boolean connect = ch.connect(addr);
- if (log.isDebugEnabled())
- log.debug(
- "Handshake timed out (will retry with increased timeout) [timeout=" + connTimeout0 +
- ", addr=" + addr + ", err=" + e + ']');
+ IgniteInClosure<IgniteInternalFuture<GridNioSession>> lsnr0 = new IgniteInClosure<IgniteInternalFuture<GridNioSession>>() {
+ @Override public void apply(final IgniteInternalFuture<GridNioSession> fut) {
+ GridNioSession ses = null;
- if (attempt == reconCnt || connTimeout0 > maxConnTimeout) {
- if (log.isDebugEnabled())
- log.debug("Handshake timedout (will stop attempts to perform the handshake) " +
- "[timeout=" + connTimeout0 + ", maxConnTimeout=" + maxConnTimeout +
- ", attempt=" + attempt + ", reconCnt=" + reconCnt +
- ", err=" + e.getMessage() + ", addr=" + addr + ']');
+ try {
+ ses = fut.get();
- if (errs == null)
- errs = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
- "Make sure that each ComputeTask and cache Transaction has a timeout set " +
- "in order to prevent parties from waiting forever in case of network issues " +
- "[nodeId=" + node.id() + ", addrs=" + addrs + ']');
+ boolean canceled = connTimeoutObj.cancel();
- errs.addSuppressed(new IgniteCheckedException("Failed to connect to address: " + addr, e));
+ if (canceled)
+ removeTimeoutObject(connTimeoutObj);
+ else {
+ final GridNioSession ses0 = ses;
- break;
- }
- else {
- attempt++;
+ Runnable clo = new Runnable() {
+ @Override public void run() {
+ GridNioFuture<Boolean> fut = nioSrvr.close(ses0);
- connTimeout0 *= 2;
+ final SocketTimeoutException e = new SocketTimeoutException("Connect timed " +
+ "(consider increasing 'connTimeout' configuration property) [addr=" +
+ currAddr + ", connTimeout=" + connTimeout + ']');
- // Continue loop.
- }
- }
- catch (Exception e) {
- if (client != null) {
- client.forceClose();
+ fut.listen(new IgniteInClosure<IgniteInternalFuture<Boolean>>() {
+ @Override public void apply(IgniteInternalFuture<Boolean> fut0) {
+ Runnable clo = new Runnable() {
+ @Override public void run() {
+ onError(e);
+ }
+ };
- client = null;
- }
+ SessionInfo sesInfo = new SessionInfo(null, SessionState.RETRY, clo);
- onException("Client creation failed [addr=" + addr + ", err=" + e + ']', e);
+ commWorker.addSessionStateChangeRequest(sesInfo);
+ }
+ });
+ }
+ };
- if (log.isDebugEnabled())
- log.debug("Client creation failed [addr=" + addr + ", err=" + e + ']');
+ commWorker.addSessionStateChangeRequest(new SessionInfo(null, SessionState.RETRY, clo));
- boolean failureDetThrReached = timeoutHelper.checkFailureTimeoutReached(e);
+ return;
+ }
- if (failureDetThrReached)
- LT.warn(log, "Connect timed out (consider increasing 'failureDetectionTimeout' " +
- "configuration property) [addr=" + addr + ", failureDetectionTimeout=" +
- failureDetectionTimeout() + ']');
- else if (X.hasCause(e, SocketTimeoutException.class))
- LT.warn(log, "Connect timed out (consider increasing 'connTimeout' " +
- "configuration property) [addr=" + addr + ", connTimeout=" + connTimeout + ']');
+ int timeoutChunk1 = (int) timeoutHelper.nextTimeoutChunk(connTimeout);
- if (errs == null)
- errs = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
- "Make sure that each ComputeTask and cache Transaction has a timeout set " +
- "in order to prevent parties from waiting forever in case of network issues " +
- "[nodeId=" + node.id() + ", addrs=" + addrs + ']');
+ long time = U.currentTimeMillis() + timeoutChunk1 * (1L << attempt0);
- errs.addSuppressed(new IgniteCheckedException("Failed to connect to address: " + addr, e));
+ HandshakeTimeoutObject<SocketChannel> handshakeTimeoutObj =
+ new HandshakeTimeoutObject<>(ch, TcpClientFuture.this, time);
- // Reconnect for the second time, if connection is not established.
- if (!failureDetThrReached && connectAttempts < 2 &&
- (e instanceof ConnectException || X.hasCause(e, ConnectException.class))) {
- connectAttempts++;
+ ctx.handshakeTimeoutObj = handshakeTimeoutObj;
- continue;
- }
+ addTimeoutObject(handshakeTimeoutObj);
+ }
+ catch (final IgniteSpiOperationTimeoutException e) {
+ assert ses != null;
- break;
- }
- }
+ final GridNioSession ses0 = ses;
- if (conn)
- break;
- }
+ commWorker.addSessionStateChangeRequest(new SessionInfo(null, SessionState.RETRY, new Runnable() {
+ @Override public void run() {
+ GridNioFuture<Boolean> closeFut = nioSrvr.close(ses0);
- if (client == null) {
- assert errs != null;
+ closeFut.listen(new IgniteInClosure<IgniteInternalFuture<Boolean>>() {
+ @Override public void apply(IgniteInternalFuture<Boolean> fut0) {
+ Runnable clo = new Runnable() {
+ @Override public void run() {
+ onError(e);
+ }
+ };
- if (X.hasCause(errs, ConnectException.class))
- LT.warn(log, "Failed to connect to a remote node " +
- "(make sure that destination node is alive and " +
- "operating system firewall is disabled on local and remote hosts) " +
- "[addrs=" + addrs + ']');
+ SessionInfo sesInfo = new SessionInfo(null, SessionState.RETRY, clo);
- if (getSpiContext().node(node.id()) != null && (CU.clientNode(node) || !CU.clientNode(getLocalNode())) &&
- X.hasCause(errs, ConnectException.class, SocketTimeoutException.class, HandshakeTimeoutException.class,
- IgniteSpiOperationTimeoutException.class)) {
- LT.warn(log, "TcpCommunicationSpi failed to establish connection to node, node will be dropped from " +
- "cluster [" +
- "rmtNode=" + node +
- ", err=" + errs +
- ", connectErrs=" + Arrays.toString(errs.getSuppressed()) + ']');
+ commWorker.addSessionStateChangeRequest(sesInfo);
+ }
+ });
+ }
+ }));
+ }
+ catch (IgniteCheckedException e) {
+ connTimeoutObj.cancel();
- getSpiContext().failNode(node.id(), "TcpCommunicationSpi failed to establish connection to node [" +
- "rmtNode=" + node +
- ", errs=" + errs +
- ", connectErrs=" + Arrays.toString(errs.getSuppressed()) + ']');
- }
+ removeTimeoutObject(connTimeoutObj);
- throw errs;
+ recoveryDesc.release();
+
+ onError(e);
+ }
+ }
+ };
+
+ nioSrvr.createSession(ch, meta, !connect, lsnr0);
+ }
+ catch (Exception e) {
+ onDone(e);
+ }
}
- return client;
- }
+ /**
+ * @param e Exception.
+ */
+ void onError(Exception e) {
+ if (e instanceof HandshakeTimeoutException || e instanceof IgniteSpiOperationTimeoutException) {
+ if (failureDetectionTimeoutEnabled() && (e instanceof HandshakeTimeoutException ||
+ timeoutHelper.checkFailureTimeoutReached(e))) {
- /**
- * Performs handshake in timeout-safe way.
- *
- * @param client Client.
- * @param recovery Recovery descriptor if use recovery handshake, otherwise {@code null}.
- * @param rmtNodeId Remote node.
- * @param timeout Timeout for handshake.
- * @param sslMeta Session meta.
- * @param handshakeConnIdx Non null connection index if need send it in handshake.
- * @throws IgniteCheckedException If handshake failed or wasn't completed withing timeout.
- * @return Handshake response.
- */
- @SuppressWarnings("ThrowFromFinallyBlock")
- private <T> long safeHandshake(
- T client,
- @Nullable GridNioRecoveryDescriptor recovery,
- UUID rmtNodeId,
- long timeout,
- GridSslMeta sslMeta,
- @Nullable Integer handshakeConnIdx
- ) throws IgniteCheckedException {
- HandshakeTimeoutObject<T> obj = new HandshakeTimeoutObject<>(client, U.currentTimeMillis() + timeout);
+ String msg = "Handshake timed out (failure detection timeout is reached) " +
+ "[failureDetectionTimeout=" + failureDetectionTimeout() + ", addr=" + currAddr + ']';
- addTimeoutObject(obj);
+ onException(msg, e);
- long rcvCnt = 0;
+ if (log.isDebugEnabled())
+ log.debug(msg);
- try {
- if (client instanceof GridCommunicationClient)
- ((GridCommunicationClient)client).doHandshake(new HandshakeClosure(rmtNodeId));
- else {
- SocketChannel ch = (SocketChannel)client;
+ if (err == null)
+ err = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
+ "Make sure that each ComputeTask and cache Transaction has a timeout set " +
+ "in order to prevent parties from waiting forever in case of network issues " +
+ "[nodeId=" + node.id() + ", addrs=" + addrs + ']');
- boolean success = false;
+ err.addSuppressed(new IgniteCheckedException("Failed to connect to address: " + currAddr, e));
- try {
- BlockingSslHandler sslHnd = null;
+ tryConnect(true);
- ByteBuffer buf;
+ return;
+ }
- if (isSslEnabled()) {
- assert sslMeta != null;
+ assert !failureDetectionTimeoutEnabled();
- sslHnd = new BlockingSslHandler(sslMeta.sslEngine(), ch, directBuf, ByteOrder.nativeOrder(), log);
+ long connTimeout0 = connTimeout * attempt;
- if (!sslHnd.handshake())
- throw new IgniteCheckedException("SSL handshake is not completed.");
+ onException("Handshake timed out (will retry with increased timeout) [timeout=" + connTimeout +
+ ", addr=" + currAddr + ']', e);
- ByteBuffer handBuff = sslHnd.applicationBuffer();
+ if (log.isDebugEnabled())
+ log.debug(
+ "Handshake timed out (will retry with increased timeout) [timeout=" + connTimeout +
+ ", addr=" + currAddr + ", err=" + e + ']');
- if (handBuff.remaining() < NodeIdMessage.MESSAGE_FULL_SIZE) {
- buf = ByteBuffer.allocate(1000);
+ if (attempt == reconCnt || connTimeout0 > maxConnTimeout) {
+ if (log.isDebugEnabled())
+ log.debug("Handshake timedout (will stop attempts to perform the handshake) " +
+ "[timeout=" + connTimeout0 + ", maxConnTimeout=" + maxConnTimeout +
+ ", attempt=" + attempt + ", reconCnt=" + reconCnt +
+ ", err=" + e.getMessage() + ", addr=" + currAddr + ']');
- int read = ch.read(buf);
+ if (err == null)
+ err = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
+ "Make sure that each ComputeTask and cache Transaction has a timeout set " +
+ "in order to prevent parties from waiting forever in case of network issues " +
+ "[nodeId=" + node.id() + ", addrs=" + addrs + ']');
- if (read == -1)
- throw new IgniteCheckedException("Failed to read remote node ID (connection closed).");
+ err.addSuppressed(new IgniteCheckedException("Failed to connect to address: " + currAddr, e));
- buf.flip();
+ tryConnect(true);
+ }
+ else {
+ attempt++;
- buf = sslHnd.decode(buf);
- }
- else
- buf = handBuff;
- }
- else {
- buf = ByteBuffer.allocate(NodeIdMessage.MESSAGE_FULL_SIZE);
+ tryConnect(false); // Reconnection on handshake timeout.
+ }
+ }
+ else {
+ onException("Client creation failed [addr=" + currAddr + ", err=" + e + ']', e);
- for (int i = 0; i < NodeIdMessage.MESSAGE_FULL_SIZE; ) {
- int read = ch.read(buf);
+ if (log.isDebugEnabled())
+ log.debug("Client creation failed [addr=" + currAddr + ", err=" + e + ']');
- if (read == -1)
- throw new IgniteCheckedException("Failed to read remote node ID (connection closed).");
+ boolean failureDetThrReached = timeoutHelper.checkFailureTimeoutReached(e);
- i += read;
- }
- }
+ if (failureDetThrReached)
+ LT.warn(log, "Connect timed out (consider increasing 'failureDetectionTimeout' " +
+ "configuration property) [addr=" + currAddr + ", failureDetectionTimeout=" +
+ failureDetectionTimeout() + ']');
+ else if (X.hasCause(e, SocketTimeoutException.class))
+ LT.warn(log, "Connect timed out (consider increasing 'connTimeout' " +
+ "configuration property) [addr=" + currAddr + ", connTimeout=" + connTimeout + ']');
- UUID rmtNodeId0 = U.bytesToUuid(buf.array(), Message.DIRECT_TYPE_SIZE);
+ if (err == null)
+ err = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
+ "Make sure that each ComputeTask and GridCacheTransaction has a timeout set " +
+ "in order to prevent parties from waiting forever in case of network issues " +
+ "[nodeId=" + node.id() + ", addrs=" + addrs + ']');
- if (!rmtNodeId.equals(rmtNodeId0))
- throw new IgniteCheckedException("Remote node ID is not as expected [expected=" + rmtNodeId +
- ", rcvd=" + rmtNodeId0 + ']');
- else if (log.isDebugEnabled())
- log.debug("Received remote node ID: " + rmtNodeId0);
+ err.addSuppressed(new IgniteCheckedException("Failed to connect to address: " + currAddr, e));
- if (isSslEnabled()) {
- assert sslHnd != null;
+ // Reconnect for the second time, if connection is not established.
+ int connectAttempts0;
- ch.write(sslHnd.encrypt(ByteBuffer.wrap(U.IGNITE_HEADER)));
- }
- else
- ch.write(ByteBuffer.wrap(U.IGNITE_HEADER));
+ if (!failureDetThrReached && (connectAttempts0 = connectAttempts) <= 2 &&
+ (e instanceof SocketTimeoutException || X.hasCause(e, SocketTimeoutException.class))) {
+ connectAttempts = connectAttempts0 + 1;
- ClusterNode locNode = getLocalNode();
+ tryConnect(false);
- if (locNode == null)
- throw new IgniteCheckedException("Local node has not been started or " +
- "fully initialized [isStopping=" + getSpiContext().isStopping() + ']');
+ return;
+ }
- if (recovery != null) {
- HandshakeMessage msg;
+ tryConnect(true);
+ }
- int msgSize = HandshakeMessage.MESSAGE_FULL_SIZE;
+ onDone(e);
+ }
- if (handshakeConnIdx != null) {
- msg = new HandshakeMessage2(locNode.id(),
- recovery.incrementConnectCount(),
- recovery.received(),
- handshakeConnIdx);
+ /**
+ *
+ */
+ private Collection<InetSocketAddress> addrs() throws IgniteCheckedException {
+ Collection<String> rmtAddrs0 = node.attribute(createSpiAttributeName(ATTR_ADDRS));
+ Collection<String> rmtHostNames0 = node.attribute(createSpiAttributeName(ATTR_HOST_NAMES));
+ Integer boundPort = node.attribute(createSpiAttributeName(ATTR_PORT));
+ Collection<InetSocketAddress> extAddrs = node.attribute(createSpiAttributeName(ATTR_EXT_ADDRS));
- msgSize += 4;
- }
- else {
- msg = new HandshakeMessage(locNode.id(),
- recovery.incrementConnectCount(),
- recovery.received());
- }
+ boolean isRmtAddrsExist = (!F.isEmpty(rmtAddrs0) && boundPort != null);
+ boolean isExtAddrsExist = !F.isEmpty(extAddrs);
- if (log.isDebugEnabled())
- log.debug("Write handshake message [rmtNode=" + rmtNodeId + ", msg=" + msg + ']');
+ if (!isRmtAddrsExist && !isExtAddrsExist)
+ throw new IgniteCheckedException("Failed to send message to the destination node. Node doesn't have any " +
+ "TCP communication addresses or mapped external addresses. Check configuration and make sure " +
+ "that you use the same communication SPI on all nodes. Remote node id: " + node.id());
- buf = ByteBuffer.allocate(msgSize);
+ LinkedHashSet<InetSocketAddress> addrs;
- buf.order(ByteOrder.nativeOrder());
+ // Try to connect first on bound addresses.
+ if (isRmtAddrsExist) {
+ List<InetSocketAddress> addrs0 = new ArrayList<>(U.toSocketAddresses(rmtAddrs0, rmtHostNames0, boundPort));
- boolean written = msg.writeTo(buf, null);
+ boolean sameHost = U.sameMacs(getSpiContext().localNode(), node);
- assert written;
+ Collections.sort(addrs0, U.inetAddressesComparator(sameHost));
- buf.flip();
+ addrs = new LinkedHashSet<>(addrs0);
+ }
+ else
+ addrs = new LinkedHashSet<>();
- if (isSslEnabled()) {
- assert sslHnd != null;
+ // Then on mapped external addresses.
+ if (isExtAddrsExist)
+ addrs.addAll(extAddrs);
- ch.write(sslHnd.encrypt(buf));
- }
- else
- ch.write(buf);
- }
- else {
- if (isSslEnabled()) {
- assert sslHnd != null;
+ Set<InetAddress> allInetAddrs = U.newHashSet(addrs.size());
- ch.write(sslHnd.encrypt(ByteBuffer.wrap(nodeIdMessage().nodeIdBytesWithType)));
- }
- else
- ch.write(ByteBuffer.wrap(nodeIdMessage().nodeIdBytesWithType));
- }
+ for (InetSocketAddress addr : addrs)
+ allInetAddrs.add(addr.getAddress());
- if (recovery != null) {
- if (log.isDebugEnabled())
- log.debug("Waiting for handshake [rmtNode=" + rmtNodeId + ']');
+ List<InetAddress> reachableInetAddrs = U.filterReachable(allInetAddrs);
+
+ if (reachableInetAddrs.size() < allInetAddrs.size()) {
+ LinkedHashSet<InetSocketAddress> addrs0 = U.newLinkedHashSet(addrs.size());
- if (isSslEnabled()) {
- assert sslHnd != null;
+ for (InetSocketAddress addr : addrs) {
+ if (reachableInetAddrs.contains(addr.getAddress()))
+ addrs0.add(addr);
+ }
+ for (InetSocketAddress addr : addrs) {
+ if (!reachableInetAddrs.contains(addr.getAddress()))
+ addrs0.add(addr);
+ }
- buf = ByteBuffer.allocate(1000);
+ addrs = addrs0;
+ }
- ByteBuffer decode = null;
+ if (log.isDebugEnabled())
+ log.debug("Addresses to connect for node [rmtNode=" + node.id() + ", addrs=" + addrs.toString() + ']');
- buf.order(ByteOrder.nativeOrder());
+ return addrs;
+ }
- for (int i = 0; i < RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE; ) {
- int read = ch.read(buf);
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(TcpClientFuture.class, this);
+ }
+ }
- if (read == -1)
- throw new IgniteCheckedException("Failed to read remote node recovery handshake " +
- "(connection closed).");
+ /**
+ * Performs handshake in timeout-safe way.
+ *
+ * @param client Client.
+ * @param rmtNodeId Remote node.
+ * @param timeout Timeout for handshake.
+ * @throws IgniteCheckedException If handshake failed or wasn't completed withing timeout.
+ * @return Handshake response.
+ */
+ @SuppressWarnings("ThrowFromFinallyBlock")
+ private <T> long safeHandshake(T client, UUID rmtNodeId, long timeout) throws IgniteCheckedException {
+ HandshakeTimeoutObject<T> obj = new HandshakeTimeoutObject<>(client, null, U.currentTimeMillis() + timeout);
- buf.flip();
+ addTimeoutObject(obj);
- decode = sslHnd.decode(buf);
+ long rcvCnt = 0;
- i += decode.remaining();
+ try {
+ if (client instanceof GridCommunicationClient)
+ ((GridCommunicationClient)client).doHandshake(new HandshakeClosure(rmtNodeId));
+ else {
+ SocketChannel ch = (SocketChannel)client;
- buf.clear();
- }
+ boolean success = false;
- rcvCnt = decode.getLong(Message.DIRECT_TYPE_SIZE);
+ try {
+ ByteBuffer buf;
- if (decode.limit() > RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE) {
- decode.position(RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE);
+ buf = ByteBuffer.allocate(17);
- sslMeta.decodedBuffer(decode);
- }
+ for (int i = 0; i < 17; ) {
+ int read = ch.read(buf);
- ByteBuffer inBuf = sslHnd.inputBuffer();
+ if (read == -1)
+ throw new IgniteCheckedException("Failed to read remote node ID (connection closed).");
- if (inBuf.position() > 0)
- sslMeta.encodedBuffer(inBuf);
- }
- else {
- buf = ByteBuffer.allocate(RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE);
+ i += read;
+ }
- buf.order(ByteOrder.nativeOrder());
+ UUID rmtNodeId0 = U.bytesToUuid(buf.array(), Message.DIRECT_TYPE_SIZE);
- for (int i = 0; i < RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE; ) {
- int read = ch.read(buf);
+ if (!rmtNodeId.equals(rmtNodeId0))
+ throw new IgniteCheckedException("Remote node ID is not as expected [expected=" + rmtNodeId +
+ ", rcvd=" + rmtNodeId0 + ']');
+ else if (log.isDebugEnabled())
+ log.debug("Received remote node ID: " + rmtNodeId0);
- if (read == -1)
- throw new IgniteCheckedException("Failed to read remote node recovery handshake " +
- "(connection closed).");
+ ch.write(ByteBuffer.wrap(U.IGNITE_HEADER));
- i += read;
- }
+ ClusterNode locNode = getLocalNode();
- rcvCnt = buf.getLong(Message.DIRECT_TYPE_SIZE);
- }
+ if (locNode == null)
+ throw new IgniteCheckedException("Local node has not been started or " +
+ "fully initialized [isStopping=" + getSpiContext().isStopping() + ']');
- if (log.isDebugEnabled())
- log.debug("Received handshake message [rmtNode=" + rmtNodeId + ", rcvCnt=" + rcvCnt + ']');
+ ch.write(ByteBuffer.wrap(nodeIdMessage().nodeIdBytesWithType));
- if (rcvCnt == -1) {
- if (log.isDebugEnabled())
- log.debug("Connection rejected, will retry client creation [rmtNode=" + rmtNodeId + ']');
- }
- else
- success = true;
- }
- else
- success = true;
+ success = true;
}
catch (IOException e) {
if (log.isDebugEnabled())
@@ -3619,7 +3936,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
*/
private class CommunicationWorker extends IgniteSpiThread {
/** */
- private final BlockingQueue<DisconnectedSessionInfo> q = new LinkedBlockingQueue<>();
+ private final BlockingQueue<SessionInfo> q = new LinkedBlockingQueue<>();
/**
* @param igniteInstanceName Ignite instance name.
@@ -3634,10 +3951,56 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
log.debug("Tcp communication worker has been started.");
while (!isInterrupted()) {
- DisconnectedSessionInfo disconnectData = q.poll(idleConnTimeout, TimeUnit.MILLISECONDS);
+ SessionInfo sesInfo = q.poll(idleConnTimeout, TimeUnit.MILLISECONDS);
+
+ if (sesInfo != null) {
+ ConnectContext ctx;
+
+ TcpClientFuture clientFut;
+
+ switch (sesInfo.state) {
+ case RECONNECT:
+ processDisconnect(sesInfo);
+
+ break;
+
+ case RETRY:
+ Runnable clo = sesInfo.clo;
+
+ assert clo != null;
- if (disconnectData != null)
- processDisconnect(disconnectData);
+ clo.run();
+
+ break;
+
+ case READY:
+ ctx = sesInfo.ses.meta(CONN_CTX_META_KEY);
+
+ assert ctx != null;
+ assert ctx.tcpClientFut != null;
+
+ GridTcpNioCommunicationClient client =
+
<TRUNCATED>
[5/9] ignite git commit: IGNITE-4302 - Use custom messages to
exchange binary metadata - Fixes #1655.
Posted by ag...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataRequestMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataRequestMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataRequestMessage.java
new file mode 100644
index 0000000..87bf805
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataRequestMessage.java
@@ -0,0 +1,122 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * As {@link DiscoveryCustomMessage} messages are delivered to client nodes asynchronously
+ * it is possible that server nodes are allowed to send to clients some BinaryObjects clients don't have metadata for.
+ *
+ * When client detects obsolete metadata (by checking if current version of metadata has schemaId)
+ * it requests up-to-date metadata using communication SPI.
+ *
+ * API to make a request is provided by {@link BinaryMetadataTransport#requestUpToDateMetadata(int)} method.
+ */
+public class MetadataRequestMessage implements Message {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private int typeId;
+
+ /**
+ * Default constructor.
+ */
+ public MetadataRequestMessage() {
+ //No-op.
+ }
+
+ /**
+ * @param typeId Type ID.
+ */
+ MetadataRequestMessage(int typeId) {
+ this.typeId = typeId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 0:
+ if (!writer.writeInt("typeId", typeId))
+ return false;
+
+ writer.incrementState();
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ if (!reader.beforeMessageRead())
+ return false;
+
+ switch (reader.state()) {
+ case 0:
+ typeId = reader.readInt("typeId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+ }
+
+ return reader.afterMessageRead(MetadataRequestMessage.class);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 80;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAckReceived() {
+ //No-op.
+ }
+
+ /** */
+ public int typeId() {
+ return typeId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MetadataRequestMessage.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataResponseMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataResponseMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataResponseMessage.java
new file mode 100644
index 0000000..bb3d3a3
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataResponseMessage.java
@@ -0,0 +1,195 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * Carries latest version of metadata to client as a response for {@link MetadataRequestMessage}.
+ */
+public class MetadataResponseMessage implements Message {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private int typeId;
+
+ /** */
+ private byte[] binaryMetadataBytes;
+
+ /** */
+ private ClientResponseStatus status;
+
+ /** */
+ public MetadataResponseMessage() {
+ // No-op.
+ }
+
+ /**
+ * @param typeId Type id.
+ */
+ MetadataResponseMessage(int typeId) {
+ this.typeId = typeId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 0:
+ if (!writer.writeInt("typeId", typeId))
+ return false;
+
+ writer.incrementState();
+
+ case 1:
+ if (!writer.writeInt("status", status.ordinal()))
+ return false;
+
+ writer.incrementState();
+
+ case 2:
+ if (!writer.writeByteArray("binMetaBytes", binaryMetadataBytes))
+ return false;
+
+ writer.incrementState();
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ if (!reader.beforeMessageRead())
+ return false;
+
+ switch (reader.state()) {
+ case 0:
+ typeId = reader.readInt("typeId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 1:
+ status = ClientResponseStatus.values()[reader.readInt("status")];
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 2:
+ binaryMetadataBytes = reader.readByteArray("binMetaBytes");
+
+ if (!reader.isLastRead())
+ return false;
+ }
+
+ return reader.afterMessageRead(MetadataResponseMessage.class);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 81;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 3;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAckReceived() {
+ // No-op.
+
+ }
+
+ /**
+ * @param bytes Binary metadata bytes.
+ */
+ void binaryMetadataBytes(byte[] bytes) {
+ if (bytes != null)
+ status = ClientResponseStatus.METADATA_FOUND;
+ else
+ status = ClientResponseStatus.METADATA_NOT_FOUND;
+
+ binaryMetadataBytes = bytes;
+ }
+
+ /**
+ * Marks message if any exception happened during preparing response.
+ */
+ void markErrorOnRequest() {
+ status = ClientResponseStatus.ERROR;
+ }
+
+ /**
+ * @return Type ID.
+ */
+ int typeId() {
+ return typeId;
+ }
+
+ /**
+ * @return Marshalled BinaryMetadata.
+ */
+ byte[] binaryMetadataBytes() {
+ return binaryMetadataBytes;
+ }
+
+ /**
+ * @return {@code true} if metadata was not found on server node replied with the response.
+ */
+ boolean metadataNotFound() {
+ return status == ClientResponseStatus.METADATA_NOT_FOUND;
+ }
+
+ /**
+ * Response statuses enum.
+ */
+ private enum ClientResponseStatus {
+ /** */
+ METADATA_FOUND,
+
+ /** */
+ METADATA_NOT_FOUND,
+
+ /** */
+ ERROR
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MetadataResponseMessage.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java
new file mode 100644
index 0000000..ef5370e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java
@@ -0,0 +1,96 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Acknowledge message for {@link MetadataUpdateProposedMessage}: see its javadoc for detailed description of protocol.
+ *
+ * As discovery messaging doesn't guarantee that message makes only one pass across the cluster
+ * <b>MetadataUpdateAcceptedMessage</b> enables to mark it as duplicated so other nodes won't process it but skip.
+ */
+public class MetadataUpdateAcceptedMessage implements DiscoveryCustomMessage {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final IgniteUuid id = IgniteUuid.randomUuid();
+
+ /** */
+ private final int typeId;
+
+ /** */
+ private final int acceptedVer;
+
+ /** */
+ private boolean duplicated;
+
+ /**
+ * @param typeId Type id.
+ * @param acceptedVer Accepted version.
+ */
+ MetadataUpdateAcceptedMessage(int typeId, int acceptedVer) {
+ this.typeId = typeId;
+ this.acceptedVer = acceptedVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid id() {
+ return id;
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public DiscoveryCustomMessage ackMessage() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isMutable() {
+ return true;
+ }
+
+ /** */
+ int acceptedVersion() {
+ return acceptedVer;
+ }
+
+ /** */
+ public int typeId() {
+ return typeId;
+ }
+
+ /** */
+ public boolean duplicated() {
+ return duplicated;
+ }
+
+ /**
+ * @param duplicated duplicated flag.
+ */
+ public void duplicated(boolean duplicated) {
+ this.duplicated = duplicated;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MetadataUpdateAcceptedMessage.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java
new file mode 100644
index 0000000..715e668
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java
@@ -0,0 +1,224 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.util.UUID;
+import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.internal.binary.BinaryMetadata;
+import org.apache.ignite.internal.binary.BinaryMetadataHandler;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * <b>MetadataUpdateProposedMessage</b> and {@link MetadataUpdateAcceptedMessage} messages make a basis for
+ * discovery-based protocol for exchanging {@link BinaryMetadata metadata} describing objects in binary format stored in Ignite caches.
+ *
+ * All interactions with binary metadata are performed through {@link BinaryMetadataHandler}
+ * interface implemented in {@link CacheObjectBinaryProcessorImpl} processor.
+ *
+ * Protocol works as follows:
+ * <ol>
+ * <li>
+ * Each thread aiming to add/update metadata sends <b>MetadataUpdateProposedMessage</b>
+ * and blocks until receiving acknowledge or reject for proposed update.
+ * </li>
+ * <li>
+ * Coordinator node checks whether proposed update is in conflict with current version of metadata
+ * for the same typeId.
+ * In case of conflict initial <b>MetadataUpdateProposedMessage</b> is marked rejected and sent to initiator.
+ * </li>
+ * <li>
+ * If there are no conflicts on coordinator, <b>pending version</b> for metadata of this typeId is bumped up by one;
+ * <b>MetadataUpdateProposedMessage</b> with <b>pending version</b> information is sent across the cluster.
+ * </li>
+ * <li>
+ * Each node on receiving non-rejected <b>MetadataUpdateProposedMessage</b> updates <b>pending version</b>
+ * for the typeId in metadata local cache.
+ * </li>
+ * <li>
+ * When <b>MetadataUpdateProposedMessage</b> finishes pass, {@link MetadataUpdateAcceptedMessage ack} is sent.
+ * Ack has the same <b>accepted version</b> as <b>pending version</b>
+ * of initial <b>MetadataUpdateProposedMessage</b> message.
+ * </li>
+ * <li>
+ * Each node on receiving <b>MetadataUpdateAcceptedMessage</b> updates accepted version for the typeId.
+ * All threads waiting for arrival of ack with this <b>accepted version</b> are unblocked.
+ * </li>
+ * </ol>
+ *
+ * If a thread on some node decides to read metadata which has ongoing update
+ * (with <b>pending version</b> strictly greater than <b>accepted version</b>)
+ * it gets blocked until {@link MetadataUpdateAcceptedMessage} arrives with <b>accepted version</b>
+ * equals to <b>pending version</b> of this metadata to the moment when is was initially read by the thread.
+ */
+public final class MetadataUpdateProposedMessage implements DiscoveryCustomMessage {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final IgniteUuid id = IgniteUuid.randomUuid();
+
+ /** */
+ private final UUID origNodeId;
+
+ /** */
+ private BinaryMetadata metadata;
+
+ /** */
+ private final int typeId;
+
+ /** */
+ private int pendingVer;
+
+ /** */
+ private int acceptedVer;
+
+ /** */
+ private ProposalStatus status = ProposalStatus.SUCCESSFUL;
+
+ /** */
+ private BinaryObjectException err;
+
+ /**
+ * @param metadata {@link BinaryMetadata} requested to be updated.
+ * @param origNodeId ID of node requested update.
+ */
+ public MetadataUpdateProposedMessage(BinaryMetadata metadata, UUID origNodeId) {
+ assert origNodeId != null;
+ assert metadata != null;
+
+ this.origNodeId = origNodeId;
+
+ this.metadata = metadata;
+ typeId = metadata.typeId();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override public IgniteUuid id() {
+ return id;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Nullable @Override public DiscoveryCustomMessage ackMessage() {
+ return (status == ProposalStatus.SUCCESSFUL) ? new MetadataUpdateAcceptedMessage(typeId, pendingVer) : null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override public boolean isMutable() {
+ return true;
+ }
+
+ /**
+ * @param err Error caused this update to be rejected.
+ */
+ void markRejected(BinaryObjectException err) {
+ status = ProposalStatus.REJECTED;
+ this.err = err;
+ }
+
+ /**
+ *
+ */
+ boolean rejected() {
+ return status == ProposalStatus.REJECTED;
+ }
+
+ /**
+ *
+ */
+ BinaryObjectException rejectionError() {
+ return err;
+ }
+
+ /**
+ * @return Pending version.
+ */
+ int pendingVersion() {
+ return pendingVer;
+ }
+
+ /**
+ * @param pendingVer New pending version.
+ */
+ void pendingVersion(int pendingVer) {
+ this.pendingVer = pendingVer;
+ }
+
+ /**
+ *
+ */
+ int acceptedVersion() {
+ return acceptedVer;
+ }
+
+ /**
+ * @param acceptedVer Accepted version.
+ */
+ void acceptedVersion(int acceptedVer) {
+ this.acceptedVer = acceptedVer;
+ }
+
+ /**
+ *
+ */
+ UUID origNodeId() {
+ return origNodeId;
+ }
+
+ /**
+ *
+ */
+ public BinaryMetadata metadata() {
+ return metadata;
+ }
+
+ /**
+ * @param metadata Metadata.
+ */
+ public void metadata(BinaryMetadata metadata) {
+ this.metadata = metadata;
+ }
+
+ /**
+ *
+ */
+ public int typeId() {
+ return typeId;
+ }
+
+ /** */
+ private enum ProposalStatus {
+ /** */
+ SUCCESSFUL,
+
+ /** */
+ REJECTED
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MetadataUpdateProposedMessage.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateResult.java
new file mode 100644
index 0000000..6c299ab
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateResult.java
@@ -0,0 +1,96 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import org.apache.ignite.binary.BinaryObjectException;
+
+/**
+ * Represents result of metadata update or metadata read request (so it is used both by server and client nodes).
+ */
+final class MetadataUpdateResult {
+ /** */
+ private final ResultType resType;
+
+ /** */
+ private final BinaryObjectException error;
+
+ /**
+ * @param resType Response type.
+ * @param error Error.
+ */
+ private MetadataUpdateResult(ResultType resType, BinaryObjectException error) {
+ this.resType = resType;
+ this.error = error;
+ }
+
+ /**
+ *
+ */
+ boolean rejected() {
+ return resType == ResultType.REJECT;
+ }
+
+ /**
+ *
+ */
+ BinaryObjectException error() {
+ return error;
+ }
+
+ /**
+ *
+ */
+ static MetadataUpdateResult createSuccessfulResult() {
+ return new MetadataUpdateResult(ResultType.SUCCESS, null);
+ }
+
+ /**
+ * @param err Error lead to request failure.
+ */
+ static MetadataUpdateResult createFailureResult(BinaryObjectException err) {
+ assert err != null;
+
+ return new MetadataUpdateResult(ResultType.REJECT, err);
+ }
+
+ /**
+ *
+ */
+ static MetadataUpdateResult createUpdateDisabledResult() {
+ return new MetadataUpdateResult(ResultType.UPDATE_DISABLED, null);
+ }
+
+ /**
+ *
+ */
+ private enum ResultType {
+ /**
+ * If request completed successfully.
+ */
+ SUCCESS,
+
+ /**
+ * If request was rejected for any reason.
+ */
+ REJECT,
+
+ /**
+ * If request arrived at the moment when node has been stopping.
+ */
+ UPDATE_DISABLED
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index 6c8df14..d65ddf5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -138,7 +138,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
private boolean locCache;
/** */
- private transient boolean keepBinary;
+ private boolean keepBinary;
/** */
private transient ConcurrentMap<Integer, PartitionRecovery> rcvs;
@@ -1389,6 +1389,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
out.writeBoolean(sync);
out.writeBoolean(ignoreExpired);
out.writeInt(taskHash);
+ out.writeBoolean(keepBinary);
}
/** {@inheritDoc} */
@@ -1410,6 +1411,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
sync = in.readBoolean();
ignoreExpired = in.readBoolean();
taskHash = in.readInt();
+ keepBinary = in.readBoolean();
cacheId = CU.cacheId(cacheName);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
index 66c19a0..804e889 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
@@ -23,7 +23,6 @@ import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ExecutorService;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
@@ -137,10 +136,10 @@ public class GridMarshallerMappingProcessor extends GridProcessorAdapter {
/**
* Adds a listener to be notified when mapping changes.
*
- * @param mappingUpdatedListener listener for mapping updated events.
+ * @param lsnr listener for mapping updated events.
*/
- public void addMappingUpdatedListener(MappingUpdatedListener mappingUpdatedListener) {
- mappingUpdatedLsnrs.add(mappingUpdatedListener);
+ public void addMappingUpdatedListener(MappingUpdatedListener lsnr) {
+ mappingUpdatedLsnrs.add(lsnr);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
index bf44723..d37eb9f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
@@ -64,6 +64,8 @@ import org.apache.ignite.internal.processors.cache.CacheEntryImpl;
import org.apache.ignite.internal.processors.cache.CacheIteratorConverter;
import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessage;
+import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateProposedMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.query.CacheQuery;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
@@ -1492,6 +1494,9 @@ public class GridServiceProcessor extends GridProcessorAdapter {
else
return;
+ if (msg instanceof MetadataUpdateProposedMessage || msg instanceof MetadataUpdateAcceptedMessage)
+ return;
+
topVer = ((DiscoveryCustomEvent)evt).affinityTopologyVersion();
}
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
index e533a70..84e37ca 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
@@ -97,6 +97,8 @@ public class GridTaskExecutionSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testJobIdCollision() throws Exception {
+ fail("Test refactoring is needed: https://issues.apache.org/jira/browse/IGNITE-4706");
+
long locId = IgniteUuid.lastLocalId();
ArrayList<IgniteFuture<Object>> futs = new ArrayList<>(2016);
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/test/java/org/apache/ignite/internal/binary/TestCachingMetadataHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/binary/TestCachingMetadataHandler.java b/modules/core/src/test/java/org/apache/ignite/internal/binary/TestCachingMetadataHandler.java
index 0f48961..704c8f3 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/binary/TestCachingMetadataHandler.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/binary/TestCachingMetadataHandler.java
@@ -42,4 +42,9 @@ public class TestCachingMetadataHandler implements BinaryMetadataHandler {
@Override public BinaryType metadata(int typeId) throws BinaryObjectException {
return metas.get(typeId);
}
+
+ /** {@inheritDoc} */
+ @Override public BinaryType metadata(int typeId, int schemaId) throws BinaryObjectException {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatesFlowTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatesFlowTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatesFlowTest.java
new file mode 100644
index 0000000..b76279d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatesFlowTest.java
@@ -0,0 +1,592 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.query.CacheQueryEntryEvent;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cluster.ClusterGroup;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
+import org.apache.ignite.internal.binary.BinaryObjectImpl;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
+import org.apache.ignite.spi.discovery.DiscoverySpiListener;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.GridTestUtils.DiscoveryHook;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class BinaryMetadataUpdatesFlowTest extends GridCommonAbstractTest {
+ /** */
+ private static final String SEQ_NUM_FLD = "f0";
+
+ /** */
+ protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private volatile boolean clientMode;
+
+ /** */
+ private volatile boolean applyDiscoveryHook;
+
+ /** */
+ private volatile DiscoveryHook discoveryHook;
+
+ /** */
+ private static final int UPDATES_COUNT = 5_000;
+
+ /** */
+ private static final int RESTART_DELAY = 3_000;
+
+ /** */
+ private final Queue<BinaryUpdateDescription> updatesQueue = new LinkedBlockingDeque<>(UPDATES_COUNT);
+
+ /** */
+ private static volatile BlockingDeque<Integer> srvResurrectQueue = new LinkedBlockingDeque<>(1);
+
+ /** */
+ private static final CountDownLatch START_LATCH = new CountDownLatch(1);
+
+ /** */
+ private static final CountDownLatch FINISH_LATCH_NO_CLIENTS = new CountDownLatch(5);
+
+ /** */
+ private static volatile AtomicBoolean stopFlag0 = new AtomicBoolean(false);
+
+ /** */
+ private static volatile AtomicBoolean stopFlag1 = new AtomicBoolean(false);
+
+ /** */
+ private static volatile AtomicBoolean stopFlag2 = new AtomicBoolean(false);
+
+ /** */
+ private static volatile AtomicBoolean stopFlag3 = new AtomicBoolean(false);
+
+ /** */
+ private static volatile AtomicBoolean stopFlag4 = new AtomicBoolean(false);
+
+ /** */
+ private static final String BINARY_TYPE_NAME = "TestBinaryType";
+
+ /** */
+ private static final int BINARY_TYPE_ID = 708045005;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ for (int i = 0; i < UPDATES_COUNT; i++) {
+ FieldType fType = null;
+ switch (i % 4) {
+ case 0:
+ fType = FieldType.NUMBER;
+ break;
+ case 1:
+ fType = FieldType.STRING;
+ break;
+ case 2:
+ fType = FieldType.ARRAY;
+ break;
+ case 3:
+ fType = FieldType.OBJECT;
+ }
+
+ updatesQueue.add(new BinaryUpdateDescription(i, "f" + (i + 1), fType));
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setPeerClassLoadingEnabled(false);
+
+ if (applyDiscoveryHook) {
+ final DiscoveryHook hook = discoveryHook != null ? discoveryHook : new DiscoveryHook();
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi() {
+ @Override public void setListener(@Nullable DiscoverySpiListener lsnr) {
+ super.setListener(GridTestUtils.DiscoverySpiListenerWrapper.wrap(lsnr, hook));
+ }
+ };
+
+ discoSpi.setHeartbeatFrequency(1000);
+
+ cfg.setDiscoverySpi(discoSpi);
+ }
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setMarshaller(new BinaryMarshaller());
+
+ cfg.setClientMode(clientMode);
+
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(CacheMode.REPLICATED);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /**
+ * Starts new ignite node and submits computation job to it.
+ * @param idx Index.
+ * @param stopFlag Stop flag.
+ */
+ private void startComputation(int idx, AtomicBoolean stopFlag) throws Exception {
+ clientMode = false;
+
+ final IgniteEx ignite0 = startGrid(idx);
+
+ ClusterGroup cg = ignite0.cluster().forNodeId(ignite0.localNode().id());
+
+ ignite0.compute(cg).withAsync().call(new BinaryObjectAdder(ignite0, updatesQueue, 30, stopFlag));
+ }
+
+ /**
+ * @param idx Index.
+ * @param deafClient Deaf client.
+ * @param observedIds Observed ids.
+ */
+ private void startListening(int idx, boolean deafClient, Set<Integer> observedIds) throws Exception {
+ clientMode = true;
+
+ ContinuousQuery qry = new ContinuousQuery();
+
+ qry.setLocalListener(new CQListener(observedIds));
+
+ if (deafClient) {
+ applyDiscoveryHook = true;
+ discoveryHook = new DiscoveryHook() {
+ @Override public void handleDiscoveryMessage(DiscoverySpiCustomMessage msg) {
+ DiscoveryCustomMessage customMsg = msg == null ? null
+ : (DiscoveryCustomMessage) IgniteUtils.field(msg, "delegate");
+
+ if (customMsg instanceof MetadataUpdateProposedMessage) {
+ if (((MetadataUpdateProposedMessage) customMsg).typeId() == BINARY_TYPE_ID)
+ GridTestUtils.setFieldValue(customMsg, "typeId", 1);
+ }
+ else if (customMsg instanceof MetadataUpdateAcceptedMessage) {
+ if (((MetadataUpdateAcceptedMessage) customMsg).typeId() == BINARY_TYPE_ID)
+ GridTestUtils.setFieldValue(customMsg, "typeId", 1);
+ }
+ }
+ };
+
+ IgniteEx client = startGrid(idx);
+
+ client.cache(null).withKeepBinary().query(qry);
+ }
+ else {
+ applyDiscoveryHook = false;
+
+ IgniteEx client = startGrid(idx);
+
+ client.cache(null).withKeepBinary().query(qry);
+ }
+ }
+
+ /**
+ *
+ */
+ private static class CQListener implements CacheEntryUpdatedListener {
+ /** */
+ private final Set<Integer> observedIds;
+
+ /**
+ * @param observedIds
+ */
+ CQListener(Set<Integer> observedIds) {
+ this.observedIds = observedIds;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable iterable) throws CacheEntryListenerException {
+ for (Object o : iterable) {
+ if (o instanceof CacheQueryEntryEvent) {
+ CacheQueryEntryEvent e = (CacheQueryEntryEvent) o;
+
+ BinaryObjectImpl val = (BinaryObjectImpl) e.getValue();
+
+ Integer seqNum = val.field(SEQ_NUM_FLD);
+
+ observedIds.add(seqNum);
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ public void testFlowNoConflicts() throws Exception {
+ startComputation(0, stopFlag0);
+
+ startComputation(1, stopFlag1);
+
+ startComputation(2, stopFlag2);
+
+ startComputation(3, stopFlag3);
+
+ startComputation(4, stopFlag4);
+
+ Thread killer = new Thread(new ServerNodeKiller());
+ Thread resurrection = new Thread(new ServerNodeResurrection());
+ killer.setName("node-killer-thread");
+ killer.start();
+ resurrection.setName("node-resurrection-thread");
+ resurrection.start();
+
+ START_LATCH.countDown();
+
+ while (!updatesQueue.isEmpty())
+ Thread.sleep(1000);
+
+ FINISH_LATCH_NO_CLIENTS.await();
+
+ IgniteEx ignite0 = grid(0);
+
+ IgniteCache<Object, Object> cache0 = ignite0.cache(null);
+
+ int cacheEntries = cache0.size(CachePeekMode.PRIMARY);
+
+ assertTrue("Cache cannot contain more entries than were put in it;", cacheEntries <= UPDATES_COUNT);
+
+ assertEquals("There are less than expected entries, data loss occurred;", UPDATES_COUNT, cacheEntries);
+
+ killer.interrupt();
+ resurrection.interrupt();
+ }
+
+ /**
+ *
+ */
+ public void testFlowNoConflictsWithClients() throws Exception {
+ startComputation(0, stopFlag0);
+
+ startComputation(1, stopFlag1);
+
+ startComputation(2, stopFlag2);
+
+ startComputation(3, stopFlag3);
+
+ startComputation(4, stopFlag4);
+
+ final Set<Integer> deafClientObservedIds = new ConcurrentHashSet<>();
+
+ startListening(5, true, deafClientObservedIds);
+
+ final Set<Integer> regClientObservedIds = new ConcurrentHashSet<>();
+
+ startListening(6, false, regClientObservedIds);
+
+ START_LATCH.countDown();
+
+ Thread killer = new Thread(new ServerNodeKiller());
+ Thread resurrection = new Thread(new ServerNodeResurrection());
+ killer.setName("node-killer-thread");
+ killer.start();
+ resurrection.setName("node-resurrection-thread");
+ resurrection.start();
+
+ while (!updatesQueue.isEmpty())
+ Thread.sleep(1000);
+
+ killer.interrupt();
+ resurrection.interrupt();
+ }
+
+ /**
+ * Runnable responsible for stopping (gracefully) server nodes during metadata updates process.
+ */
+ private final class ServerNodeKiller implements Runnable {
+ /** {@inheritDoc} */
+ @Override public void run() {
+ Thread curr = Thread.currentThread();
+ try {
+ START_LATCH.await();
+
+ while (!curr.isInterrupted()) {
+ int idx = ThreadLocalRandom.current().nextInt(5);
+
+ AtomicBoolean stopFlag;
+
+ switch (idx) {
+ case 0:
+ stopFlag = stopFlag0;
+ break;
+ case 1:
+ stopFlag = stopFlag1;
+ break;
+ case 2:
+ stopFlag = stopFlag2;
+ break;
+ case 3:
+ stopFlag = stopFlag3;
+ break;
+ default:
+ stopFlag = stopFlag4;
+ }
+
+ stopFlag.set(true);
+
+ while (stopFlag.get())
+ Thread.sleep(10);
+
+ stopGrid(idx);
+
+ srvResurrectQueue.put(idx);
+
+ Thread.sleep(RESTART_DELAY);
+ }
+ }
+ catch (Exception ignored) {
+ // No-op.
+ }
+ }
+ }
+
+ /**
+ * {@link Runnable} object to restart nodes killed by {@link ServerNodeKiller}.
+ */
+ private final class ServerNodeResurrection implements Runnable {
+ /** {@inheritDoc} */
+ @Override public void run() {
+ Thread curr = Thread.currentThread();
+
+ try {
+ START_LATCH.await();
+
+ while (!curr.isInterrupted()) {
+ Integer idx = srvResurrectQueue.takeFirst();
+
+ AtomicBoolean stopFlag;
+
+ switch (idx) {
+ case 0:
+ stopFlag = stopFlag0;
+ break;
+ case 1:
+ stopFlag = stopFlag1;
+ break;
+ case 2:
+ stopFlag = stopFlag2;
+ break;
+ case 3:
+ stopFlag = stopFlag3;
+ break;
+ default:
+ stopFlag = stopFlag4;
+ }
+
+ clientMode = false;
+ applyDiscoveryHook = false;
+
+ startComputation(idx, stopFlag);
+ }
+ }
+ catch (Exception ignored) {
+ // No-op.
+ }
+ }
+ }
+
+ /**
+ * Instruction for node to perform <b>add new binary object</b> action on cache in <b>keepBinary</b> mode.
+ *
+ * Instruction includes id the object should be added under, new field to add to binary schema
+ * and {@link FieldType type} of the field.
+ */
+ private static final class BinaryUpdateDescription {
+ /** */
+ private int itemId;
+
+ /** */
+ private String fieldName;
+
+ /** */
+ private FieldType fieldType;
+
+ /**
+ * @param itemId Item id.
+ * @param fieldName Field name.
+ * @param fieldType Field type.
+ */
+ private BinaryUpdateDescription(int itemId, String fieldName, FieldType fieldType) {
+ this.itemId = itemId;
+ this.fieldName = fieldName;
+ this.fieldType = fieldType;
+ }
+ }
+
+ /**
+ *
+ */
+ private enum FieldType {
+ /** */
+ NUMBER,
+
+ /** */
+ STRING,
+
+ /** */
+ ARRAY,
+
+ /** */
+ OBJECT
+ }
+
+ /**
+ * Generates random number to use when creating binary object with field of numeric {@link FieldType type}.
+ */
+ private static int getNumberFieldVal() {
+ return ThreadLocalRandom.current().nextInt(100);
+ }
+
+ /**
+ * Generates random string to use when creating binary object with field of string {@link FieldType type}.
+ */
+ private static String getStringFieldVal() {
+ return "str" + (100 + ThreadLocalRandom.current().nextInt(9));
+ }
+
+ /**
+ * Generates random array to use when creating binary object with field of array {@link FieldType type}.
+ */
+ private static byte[] getArrayFieldVal() {
+ byte[] res = new byte[3];
+ ThreadLocalRandom.current().nextBytes(res);
+ return res;
+ }
+
+ /**
+ * @param builder Builder.
+ * @param desc Descriptor with parameters of BinaryObject to build.
+ * @return BinaryObject built by provided description
+ */
+ private static BinaryObject newBinaryObject(BinaryObjectBuilder builder, BinaryUpdateDescription desc) {
+ builder.setField(SEQ_NUM_FLD, desc.itemId + 1);
+
+ switch (desc.fieldType) {
+ case NUMBER:
+ builder.setField(desc.fieldName, getNumberFieldVal());
+ break;
+ case STRING:
+ builder.setField(desc.fieldName, getStringFieldVal());
+ break;
+ case ARRAY:
+ builder.setField(desc.fieldName, getArrayFieldVal());
+ break;
+ case OBJECT:
+ builder.setField(desc.fieldName, new Object());
+ }
+
+ return builder.build();
+ }
+
+ /**
+ * Compute job executed on each node in cluster which constantly adds new entries to ignite cache
+ * according to {@link BinaryUpdateDescription descriptions} it reads from shared queue.
+ */
+ private static final class BinaryObjectAdder implements IgniteCallable<Object> {
+ /** */
+ private final IgniteEx ignite;
+
+ /** */
+ private final Queue<BinaryUpdateDescription> updatesQueue;
+
+ /** */
+ private final long timeout;
+
+ /** */
+ private final AtomicBoolean stopFlag;
+
+ /**
+ * @param ignite Ignite.
+ * @param updatesQueue Updates queue.
+ * @param timeout Timeout.
+ * @param stopFlag Stop flag.
+ */
+ BinaryObjectAdder(IgniteEx ignite, Queue<BinaryUpdateDescription> updatesQueue, long timeout, AtomicBoolean stopFlag) {
+ this.ignite = ignite;
+ this.updatesQueue = updatesQueue;
+ this.timeout = timeout;
+ this.stopFlag = stopFlag;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object call() throws Exception {
+ START_LATCH.await();
+
+ IgniteCache<Object, Object> cache = ignite.cache(null).withKeepBinary();
+
+ while (!updatesQueue.isEmpty()) {
+ BinaryUpdateDescription desc = updatesQueue.poll();
+
+ BinaryObjectBuilder builder = ignite.binary().builder(BINARY_TYPE_NAME);
+
+ BinaryObject bo = newBinaryObject(builder, desc);
+
+ cache.put(desc.itemId, bo);
+
+ if (stopFlag.get())
+ break;
+ else
+ Thread.sleep(timeout);
+ }
+
+ if (updatesQueue.isEmpty())
+ FINISH_LATCH_NO_CLIENTS.countDown();
+
+ stopFlag.set(false);
+
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectMetadataExchangeMultinodeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectMetadataExchangeMultinodeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectMetadataExchangeMultinodeTest.java
new file mode 100644
index 0000000..2dcfab8
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectMetadataExchangeMultinodeTest.java
@@ -0,0 +1,463 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import junit.framework.AssertionFailedError;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cluster.ClusterGroup;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
+import org.apache.ignite.internal.managers.communication.GridIoManager;
+import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
+import org.apache.ignite.spi.discovery.DiscoverySpiListener;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.GridTestUtils.DiscoveryHook;
+import org.apache.ignite.testframework.GridTestUtils.DiscoverySpiListenerWrapper;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class GridCacheBinaryObjectMetadataExchangeMultinodeTest extends GridCommonAbstractTest {
+ /** */
+ protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private boolean clientMode;
+
+ /** */
+ private boolean applyDiscoveryHook;
+
+ /** */
+ private DiscoveryHook discoveryHook;
+
+ /** */
+ private static final String BINARY_TYPE_NAME = "TestBinaryType";
+
+ /** */
+ private static final int BINARY_TYPE_ID = 708045005;
+
+ /** */
+ private static final AtomicInteger metadataReqsCounter = new AtomicInteger(0);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ if (applyDiscoveryHook) {
+ final DiscoveryHook hook = discoveryHook != null ? discoveryHook : new DiscoveryHook();
+
+ cfg.setDiscoverySpi(new TcpDiscoverySpi() {
+ @Override public void setListener(@Nullable DiscoverySpiListener lsnr) {
+ super.setListener(DiscoverySpiListenerWrapper.wrap(lsnr, hook));
+ }
+ });
+ }
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setMarshaller(new BinaryMarshaller());
+
+ cfg.setClientMode(clientMode);
+
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(CacheMode.REPLICATED);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /**
+ *
+ */
+ private static final class ErrorHolder {
+ /** */
+ private volatile Error e;
+
+ /**
+ * @param e Exception.
+ */
+ void error(Error e) {
+ this.e = e;
+ }
+
+ /**
+ *
+ */
+ void fail() {
+ throw e;
+ }
+
+ /**
+ *
+ */
+ boolean isEmpty() {
+ return e == null;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** */
+ private static final CountDownLatch LATCH1 = new CountDownLatch(1);
+
+ /**
+ * Verifies that if thread tries to read metadata with ongoing update it gets blocked
+ * until acknowledge message arrives.
+ */
+ public void testReadRequestBlockedOnUpdatingMetadata() throws Exception {
+ applyDiscoveryHook = true;
+ discoveryHook = new DiscoveryHook() {
+ @Override public void handleDiscoveryMessage(DiscoverySpiCustomMessage msg) {
+ DiscoveryCustomMessage customMsg = msg == null ? null
+ : (DiscoveryCustomMessage) IgniteUtils.field(msg, "delegate");
+
+ if (customMsg instanceof MetadataUpdateAcceptedMessage) {
+ if (((MetadataUpdateAcceptedMessage)customMsg).typeId() == BINARY_TYPE_ID)
+ try {
+ Thread.sleep(300);
+ }
+ catch (InterruptedException ignored) {
+ // No-op.
+ }
+ }
+ }
+ };
+
+ final IgniteEx ignite0 = startGrid(0);
+
+ applyDiscoveryHook = false;
+
+ final IgniteEx ignite1 = startGrid(1);
+
+ final ErrorHolder errorHolder = new ErrorHolder();
+
+ applyDiscoveryHook = true;
+ discoveryHook = new DiscoveryHook() {
+ private volatile IgniteEx ignite;
+
+ @Override public void handleDiscoveryMessage(DiscoverySpiCustomMessage msg) {
+ DiscoveryCustomMessage customMsg = msg == null ? null
+ : (DiscoveryCustomMessage) IgniteUtils.field(msg, "delegate");
+
+ if (customMsg instanceof MetadataUpdateAcceptedMessage) {
+ MetadataUpdateAcceptedMessage acceptedMsg = (MetadataUpdateAcceptedMessage)customMsg;
+ if (acceptedMsg.typeId() == BINARY_TYPE_ID && acceptedMsg.acceptedVersion() == 2) {
+ Object binaryProc = U.field(ignite.context(), "cacheObjProc");
+ Object transport = U.field(binaryProc, "transport");
+
+ try {
+ Map syncMap = U.field(transport, "syncMap");
+
+ int size = syncMap.size();
+ assertEquals("unexpected size of syncMap: ", 1, size);
+
+ Object syncKey = syncMap.keySet().iterator().next();
+
+ int typeId = U.field(syncKey, "typeId");
+ assertEquals("unexpected typeId: ", BINARY_TYPE_ID, typeId);
+
+ int ver = U.field(syncKey, "ver");
+ assertEquals("unexpected pendingVersion: ", 2, ver);
+ }
+ catch (AssertionFailedError err) {
+ errorHolder.error(err);
+ }
+ }
+ }
+ }
+
+ @Override public void ignite(IgniteEx ignite) {
+ this.ignite = ignite;
+ }
+ };
+
+ final IgniteEx ignite2 = startGrid(2);
+ discoveryHook.ignite(ignite2);
+
+ ignite0.executorService().submit(new Runnable() {
+ @Override public void run() {
+ addIntField(ignite0, "f1", 101, 1);
+ }
+ }).get();
+
+ UUID id2 = ignite2.localNode().id();
+
+ ClusterGroup cg2 = ignite2.cluster().forNodeId(id2);
+
+ Future<?> fut = ignite1.executorService().submit(new Runnable() {
+ @Override public void run() {
+ LATCH1.countDown();
+ addStringField(ignite1, "f2", "str", 2);
+ }
+ });
+
+ ignite2.compute(cg2).withAsync().call(new IgniteCallable<Object>() {
+ @Override public Object call() throws Exception {
+ try {
+ LATCH1.await();
+ }
+ catch (InterruptedException ignored) {
+ // No-op.
+ }
+
+ Object fieldVal = ((BinaryObject) ignite2.cache(null).withKeepBinary().get(1)).field("f1");
+
+ return fieldVal;
+ }
+ });
+
+ fut.get();
+
+ if (!errorHolder.isEmpty())
+ errorHolder.fail();
+ }
+
+ /**
+ * Verifies that all sequential updates that don't introduce any conflicts are accepted and observed by all nodes.
+ */
+ public void testSequentialUpdatesNoConflicts() throws Exception {
+ IgniteEx ignite0 = startGrid(0);
+
+ final IgniteEx ignite1 = startGrid(1);
+
+ final String intFieldName = "f1";
+
+ ignite1.executorService().submit(new Runnable() {
+ @Override public void run() {
+ addIntField(ignite1, intFieldName, 101, 1);
+ }
+ }).get();
+
+ int fld = ((BinaryObject) ignite0.cache(null).withKeepBinary().get(1)).field(intFieldName);
+
+ assertEquals(fld, 101);
+
+ final IgniteEx ignite2 = startGrid(2);
+
+ final String strFieldName = "f2";
+
+ ignite2.executorService().submit(new Runnable() {
+ @Override public void run() {
+ addStringField(ignite2, strFieldName, "str", 2);
+ }
+ }).get();
+
+ assertEquals(((BinaryObject)ignite1.cache(null).withKeepBinary().get(2)).field(strFieldName), "str");
+ }
+
+ /**
+ * Verifies that client is able to detect obsolete metadata situation and request up-to-date from the cluster.
+ */
+ public void testClientRequestsUpToDateMetadata() throws Exception {
+ final IgniteEx ignite0 = startGrid(0);
+
+ final IgniteEx ignite1 = startGrid(1);
+
+ ignite0.executorService().submit(new Runnable() {
+ @Override public void run() {
+ addIntField(ignite0, "f1", 101, 1);
+ }
+ }).get();
+
+ final Ignite client = startDeafClient("client");
+
+ ClusterGroup clientGrp = client.cluster().forClients();
+
+ final String strVal = "strVal101";
+
+ ignite1.executorService().submit(new Runnable() {
+ @Override public void run() {
+ addStringField(ignite1, "f2", strVal, 1);
+ }
+ }).get();
+
+ String res = client.compute(clientGrp).call(new IgniteCallable<String>() {
+ @Override public String call() throws Exception {
+ return ((BinaryObject)client.cache(null).withKeepBinary().get(1)).field("f2");
+ }
+ });
+
+ assertEquals(strVal, res);
+ }
+
+ /**
+ * Verifies that client resends request for up-to-date metadata in case of failure on server received first request.
+ */
+ public void testClientRequestsUpToDateMetadataOneNodeDies() throws Exception {
+ final Ignite srv0 = startGrid(0);
+ replaceWithStoppingMappingRequestListener(((GridKernalContext)U.field(srv0, "ctx")).io(), 0);
+
+ final Ignite srv1 = startGrid(1);
+ replaceWithCountingMappingRequestListener(((GridKernalContext)U.field(srv1, "ctx")).io());
+
+ final Ignite srv2 = startGrid(2);
+ replaceWithCountingMappingRequestListener(((GridKernalContext)U.field(srv2, "ctx")).io());
+
+ final Ignite client = startDeafClient("client");
+
+ ClusterGroup clientGrp = client.cluster().forClients();
+
+ srv0.executorService().submit(new Runnable() {
+ @Override public void run() {
+ addStringField(srv0, "f2", "strVal101", 0);
+ }
+ }).get();
+
+ client.compute(clientGrp).call(new IgniteCallable<String>() {
+ @Override public String call() throws Exception {
+ return ((BinaryObject)client.cache(null).withKeepBinary().get(0)).field("f2");
+ }
+ });
+
+ assertEquals(metadataReqsCounter.get(), 2);
+ }
+
+ /**
+ * Starts client node that skips <b>MetadataUpdateProposedMessage</b> and <b>MetadataUpdateAcceptedMessage</b>
+ * messages.
+ *
+ * @param clientName name of client node.
+ */
+ private Ignite startDeafClient(String clientName) throws Exception {
+ clientMode = true;
+ applyDiscoveryHook = true;
+ discoveryHook = new DiscoveryHook() {
+ @Override public void handleDiscoveryMessage(DiscoverySpiCustomMessage msg) {
+ DiscoveryCustomMessage customMsg = msg == null ? null
+ : (DiscoveryCustomMessage) IgniteUtils.field(msg, "delegate");
+
+ if (customMsg instanceof MetadataUpdateProposedMessage) {
+ if (((MetadataUpdateProposedMessage) customMsg).typeId() == BINARY_TYPE_ID)
+ GridTestUtils.setFieldValue(customMsg, "typeId", 1);
+ }
+ else if (customMsg instanceof MetadataUpdateAcceptedMessage) {
+ if (((MetadataUpdateAcceptedMessage) customMsg).typeId() == BINARY_TYPE_ID)
+ GridTestUtils.setFieldValue(customMsg, "typeId", 1);
+ }
+ }
+ };
+
+ Ignite client = startGrid(clientName);
+
+ clientMode = false;
+ applyDiscoveryHook = false;
+
+ return client;
+ }
+
+ /**
+ *
+ */
+ private void replaceWithStoppingMappingRequestListener(GridIoManager ioMgr, final int nodeIdToStop) {
+ ioMgr.removeMessageListener(GridTopic.TOPIC_METADATA_REQ);
+
+ ioMgr.addMessageListener(GridTopic.TOPIC_METADATA_REQ, new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ new Thread(new Runnable() {
+ @Override public void run() {
+ metadataReqsCounter.incrementAndGet();
+ stopGrid(nodeIdToStop, true);
+ }
+ }).start();
+ }
+ });
+ }
+
+ /**
+ *
+ */
+ private void replaceWithCountingMappingRequestListener(GridIoManager ioMgr) {
+ GridMessageListener[] lsnrs = U.field(ioMgr, "sysLsnrs");
+
+ final GridMessageListener delegate = lsnrs[GridTopic.TOPIC_METADATA_REQ.ordinal()];
+
+ GridMessageListener wrapper = new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ metadataReqsCounter.incrementAndGet();
+ delegate.onMessage(nodeId, msg);
+ }
+ };
+
+ lsnrs[GridTopic.TOPIC_METADATA_REQ.ordinal()] = wrapper;
+ }
+
+ /**
+ * Adds field of integer type to fixed binary type.
+ *
+ * @param ignite Ignite.
+ * @param fieldName Field name.
+ * @param fieldVal Field value.
+ * @param cacheIdx Cache index.
+ */
+ private void addIntField(Ignite ignite, String fieldName, int fieldVal, int cacheIdx) {
+ BinaryObjectBuilder builder = ignite.binary().builder(BINARY_TYPE_NAME);
+
+ IgniteCache<Object, Object> cache = ignite.cache(null).withKeepBinary();
+
+ builder.setField(fieldName, fieldVal);
+
+ cache.put(cacheIdx, builder.build());
+ }
+
+ /**
+ * Adds field of String type to fixed binary type.
+ *
+ * @param ignite Ignite.
+ * @param fieldName Field name.
+ * @param fieldVal Field value.
+ * @param cacheIdx Cache index.
+ */
+ private void addStringField(Ignite ignite, String fieldName, String fieldVal, int cacheIdx) {
+ BinaryObjectBuilder builder = ignite.binary().builder(BINARY_TYPE_NAME);
+
+ IgniteCache<Object, Object> cache = ignite.cache(null).withKeepBinary();
+
+ builder.setField(fieldName, fieldVal);
+
+ cache.put(cacheIdx, builder.build());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
index 91220b2..475bf8e 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
@@ -96,6 +96,8 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
+import org.apache.ignite.spi.discovery.DiscoverySpiListener;
import org.apache.ignite.spi.swapspace.inmemory.GridTestSwapSpaceSpi;
import org.apache.ignite.ssl.SslContextFactory;
import org.apache.ignite.testframework.config.GridTestProperties;
@@ -110,6 +112,61 @@ public final class GridTestUtils {
/** Default busy wait sleep interval in milliseconds. */
public static final long DFLT_BUSYWAIT_SLEEP_INTERVAL = 200;
+
+
+ /**
+ * Hook object intervenes to discovery message handling
+ * and thus allows to make assertions or other actions like skipping certain discovery messages.
+ */
+ public static class DiscoveryHook {
+ /**
+ * @param msg Message.
+ */
+ public void handleDiscoveryMessage(DiscoverySpiCustomMessage msg) {
+ }
+
+ /**
+ * @param ignite Ignite.
+ */
+ public void ignite(IgniteEx ignite) {
+ // No-op.
+ }
+ }
+
+ /**
+ * Injects {@link DiscoveryHook} into handling logic.
+ */
+ public static final class DiscoverySpiListenerWrapper implements DiscoverySpiListener {
+ /** */
+ private final DiscoverySpiListener delegate;
+
+ /** */
+ private final DiscoveryHook hook;
+
+ /**
+ * @param delegate Delegate.
+ * @param hook Hook.
+ */
+ private DiscoverySpiListenerWrapper(DiscoverySpiListener delegate, DiscoveryHook hook) {
+ this.hook = hook;
+ this.delegate = delegate;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onDiscovery(int type, long topVer, ClusterNode node, Collection<ClusterNode> topSnapshot, @Nullable Map<Long, Collection<ClusterNode>> topHist, @Nullable DiscoverySpiCustomMessage spiCustomMsg) {
+ hook.handleDiscoveryMessage(spiCustomMsg);
+ delegate.onDiscovery(type, topVer, node, topSnapshot, topHist, spiCustomMsg);
+ }
+
+ /**
+ * @param delegate Delegate.
+ * @param discoveryHook Discovery hook.
+ */
+ public static DiscoverySpiListener wrap(DiscoverySpiListener delegate, DiscoveryHook discoveryHook) {
+ return new DiscoverySpiListenerWrapper(delegate, discoveryHook);
+ }
+ }
+
/** */
private static final Map<Class<?>, String> addrs = new HashMap<>();
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
index 1a348e0..a462f90 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
@@ -54,6 +54,8 @@ import org.apache.ignite.internal.binary.noncompact.BinaryObjectBuilderNonCompac
import org.apache.ignite.internal.binary.streams.BinaryHeapStreamByteOrderSelfTest;
import org.apache.ignite.internal.binary.streams.BinaryOffheapStreamByteOrderSelfTest;
import org.apache.ignite.internal.processors.cache.BinaryObjectOffHeapUnswapTemporaryTest;
+import org.apache.ignite.internal.processors.cache.binary.BinaryMetadataUpdatesFlowTest;
+import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryObjectMetadataExchangeMultinodeTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryObjectUserClassloaderSelfTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryStoreBinariesDefaultMappersSelfTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryStoreBinariesSimpleNameMappersSelfTest;
@@ -147,6 +149,8 @@ public class IgniteBinaryObjectsTestSuite extends TestSuite {
suite.addTestSuite(GridCacheBinaryStoreBinariesSimpleNameMappersSelfTest.class);
suite.addTestSuite(GridCacheClientNodeBinaryObjectMetadataTest.class);
+ suite.addTestSuite(GridCacheBinaryObjectMetadataExchangeMultinodeTest.class);
+ suite.addTestSuite(BinaryMetadataUpdatesFlowTest.class);
suite.addTestSuite(GridCacheClientNodeBinaryObjectMetadataMultinodeTest.class);
suite.addTestSuite(IgniteBinaryMetadataUpdateChangingTopologySelfTest.class);
[6/9] ignite git commit: IGNITE-4302 - Use custom messages to
exchange binary metadata - Fixes #1655.
Posted by ag...@apache.org.
IGNITE-4302 - Use custom messages to exchange binary metadata - Fixes #1655.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/44cf1d21
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/44cf1d21
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/44cf1d21
Branch: refs/heads/ignite-4003
Commit: 44cf1d21ad4363041043ad88161a954c738f20eb
Parents: f056f2e
Author: Sergey Chugunov <se...@gmail.com>
Authored: Thu Mar 30 11:40:35 2017 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Thu Mar 30 11:40:35 2017 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/GridComponent.java | 5 +-
.../org/apache/ignite/internal/GridTopic.java | 5 +-
.../binary/BinaryCachingMetadataHandler.java | 6 +
.../ignite/internal/binary/BinaryContext.java | 10 +
.../ignite/internal/binary/BinaryMetadata.java | 27 +-
.../internal/binary/BinaryMetadataHandler.java | 18 +-
.../binary/BinaryNoopMetadataHandler.java | 5 +
.../internal/binary/BinaryReaderExImpl.java | 2 +-
.../ignite/internal/binary/BinaryUtils.java | 2 +-
.../communication/GridIoMessageFactory.java | 12 +
.../discovery/DiscoveryCustomMessage.java | 48 ++
.../eventstorage/GridEventStorageManager.java | 6 +-
.../cache/binary/BinaryMetadataHolder.java | 73 +++
.../cache/binary/BinaryMetadataTransport.java | 641 +++++++++++++++++++
.../binary/BinaryMetadataUpdatedListener.java | 29 +
.../binary/CacheObjectBinaryProcessor.java | 9 +
.../binary/CacheObjectBinaryProcessorImpl.java | 474 ++++----------
.../binary/ClientMetadataRequestFuture.java | 161 +++++
.../cache/binary/MetadataRequestMessage.java | 122 ++++
.../cache/binary/MetadataResponseMessage.java | 195 ++++++
.../binary/MetadataUpdateAcceptedMessage.java | 96 +++
.../binary/MetadataUpdateProposedMessage.java | 224 +++++++
.../cache/binary/MetadataUpdateResult.java | 96 +++
.../continuous/CacheContinuousQueryHandler.java | 4 +-
.../GridMarshallerMappingProcessor.java | 7 +-
.../service/GridServiceProcessor.java | 5 +
.../internal/GridTaskExecutionSelfTest.java | 2 +
.../binary/TestCachingMetadataHandler.java | 5 +
.../binary/BinaryMetadataUpdatesFlowTest.java | 592 +++++++++++++++++
...naryObjectMetadataExchangeMultinodeTest.java | 463 ++++++++++++++
.../ignite/testframework/GridTestUtils.java | 57 ++
.../IgniteBinaryObjectsTestSuite.java | 4 +
32 files changed, 3056 insertions(+), 349 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java b/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
index 560d7f6..dc14ae3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
@@ -49,7 +49,10 @@ public interface GridComponent {
CLUSTER_PROC,
/** */
- MARSHALLER_PROC
+ MARSHALLER_PROC,
+
+ /** */
+ BINARY_PROC
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
index 3ffc56e..6efc0d0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
@@ -103,7 +103,10 @@ public enum GridTopic {
TOPIC_MAPPING_MARSH,
/** */
- TOPIC_HADOOP_MSG;
+ TOPIC_HADOOP_MSG,
+
+ /** */
+ TOPIC_METADATA_REQ;
/** Enum values. */
private static final GridTopic[] VALS = values();
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryCachingMetadataHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryCachingMetadataHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryCachingMetadataHandler.java
index 39189f0..24a393d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryCachingMetadataHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryCachingMetadataHandler.java
@@ -67,4 +67,10 @@ public class BinaryCachingMetadataHandler implements BinaryMetadataHandler {
@Override public synchronized BinaryType metadata(int typeId) throws BinaryObjectException {
return metas.get(typeId);
}
+
+ /** {@inheritDoc} */
+ @Override public synchronized BinaryType metadata(int typeId, int schemaId) throws BinaryObjectException {
+ BinaryTypeImpl type = (BinaryTypeImpl) metas.get(typeId);
+ return type.metadata().hasSchema(schemaId) ? type : null;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
index e5b6bda..d918aa3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
@@ -1224,6 +1224,16 @@ public class BinaryContext {
/**
* @param typeId Type ID.
+ * @param schemaId Schema ID.
+ * @return Meta data.
+ * @throws BinaryObjectException In case of error.
+ */
+ public BinaryType metadata(int typeId, int schemaId) throws BinaryObjectException {
+ return metaHnd != null ? metaHnd.metadata(typeId, schemaId): null;
+ }
+
+ /**
+ * @param typeId Type ID.
* @return Affinity key field name.
*/
public String affinityKeyFieldName(int typeId) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
index d1c79f3..393d9ee 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadata.java
@@ -28,7 +28,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-
+import java.util.Set;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -60,6 +60,9 @@ public class BinaryMetadata implements Externalizable {
/** Schemas associated with type. */
private Collection<BinarySchema> schemas;
+ /** Schema IDs registered for this type */
+ private Set<Integer> schemaIds;
+
/** Whether this is enum type. */
private boolean isEnum;
@@ -89,6 +92,16 @@ public class BinaryMetadata implements Externalizable {
this.fields = fields;
this.affKeyFieldName = affKeyFieldName;
this.schemas = schemas;
+
+ if (schemas != null) {
+ schemaIds = U.newHashSet(schemas.size());
+
+ for (BinarySchema schema : schemas)
+ schemaIds.add(schema.schemaId());
+ }
+ else
+ schemaIds = Collections.emptySet();
+
this.isEnum = isEnum;
}
@@ -145,6 +158,14 @@ public class BinaryMetadata implements Externalizable {
}
/**
+ * @param schemaId Schema ID.
+ * @return {@code true} if <b>BinaryMetadata</b> instance has schema with ID specified, {@code false} otherwise.
+ */
+ public boolean hasSchema(int schemaId) {
+ return schemaIds.contains(schemaId);
+ }
+
+ /**
* @return {@code True} if this is enum type.
*/
public boolean isEnum() {
@@ -249,12 +270,16 @@ public class BinaryMetadata implements Externalizable {
else {
schemas = new ArrayList<>();
+ schemaIds = U.newHashSet(schemasSize);
+
for (int i = 0; i < schemasSize; i++) {
BinarySchema schema = new BinarySchema();
schema.readFrom(in);
schemas.add(schema);
+
+ schemaIds.add(schema.schemaId());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadataHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadataHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadataHandler.java
index 29ff7b3..748a283 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadataHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryMetadataHandler.java
@@ -28,8 +28,8 @@ public interface BinaryMetadataHandler {
* Adds meta data.
*
* @param typeId Type ID.
- * @param meta Meta data.
- * @throws org.apache.ignite.binary.BinaryObjectException In case of error.
+ * @param meta Metadata.
+ * @throws BinaryObjectException In case of error.
*/
public void addMeta(int typeId, BinaryType meta) throws BinaryObjectException;
@@ -37,8 +37,18 @@ public interface BinaryMetadataHandler {
* Gets meta data for provided type ID.
*
* @param typeId Type ID.
- * @return Meta data.
- * @throws org.apache.ignite.binary.BinaryObjectException In case of error.
+ * @return Metadata.
+ * @throws BinaryObjectException In case of error.
*/
public BinaryType metadata(int typeId) throws BinaryObjectException;
+
+ /**
+ * Gets metadata for provided type ID and schema ID.
+ *
+ * @param typeId Type ID.
+ * @param schemaId Schema ID.
+ * @return Metadata.
+ * @throws BinaryObjectException In case of error.
+ */
+ public BinaryType metadata(int typeId, int schemaId) throws BinaryObjectException;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryNoopMetadataHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryNoopMetadataHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryNoopMetadataHandler.java
index 9c0c37d..770d9c8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryNoopMetadataHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryNoopMetadataHandler.java
@@ -50,4 +50,9 @@ public class BinaryNoopMetadataHandler implements BinaryMetadataHandler {
@Override public BinaryType metadata(int typeId) throws BinaryObjectException {
return null;
}
+
+ /** {@inheritDoc} */
+ @Override public BinaryType metadata(int typeId, int schemaId) throws BinaryObjectException {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryReaderExImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryReaderExImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryReaderExImpl.java
index f851b68..d6fefe3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryReaderExImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryReaderExImpl.java
@@ -1984,7 +1984,7 @@ public class BinaryReaderExImpl implements BinaryReader, BinaryRawReaderEx, Bina
if (schema == null) {
if (fieldIdLen != BinaryUtils.FIELD_ID_LEN) {
- BinaryTypeImpl type = (BinaryTypeImpl)ctx.metadata(typeId);
+ BinaryTypeImpl type = (BinaryTypeImpl) ctx.metadata(typeId, schemaId);
if (type == null || type.metadata() == null)
throw new BinaryObjectException("Cannot find metadata for object with compact footer: " +
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
index 41ec078..c59b8b7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
@@ -1562,7 +1562,7 @@ public class BinaryUtils {
Class cls;
if (typeId != GridBinaryMarshaller.UNREGISTERED_TYPE_ID)
- cls = ctx.descriptorForTypeId(true, typeId, ldr, false).describedClass();
+ cls = ctx.descriptorForTypeId(true, typeId, ldr, true).describedClass();
else {
String clsName = doReadClassName(in);
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 7bf3de2..737d047 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -46,6 +46,8 @@ import org.apache.ignite.internal.processors.cache.GridCacheEvictionRequest;
import org.apache.ignite.internal.processors.cache.GridCacheEvictionResponse;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
+import org.apache.ignite.internal.processors.cache.binary.MetadataRequestMessage;
+import org.apache.ignite.internal.processors.cache.binary.MetadataResponseMessage;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTtlUpdateRequest;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryRequest;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryResponse;
@@ -651,6 +653,16 @@ public class GridIoMessageFactory implements MessageFactory {
break;
+ case 80:
+ msg = new MetadataRequestMessage();
+
+ break;
+
+ case 81:
+ msg = new MetadataResponseMessage();
+
+ break;
+
case 82:
msg = new JobStealingRequest();
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java
index d85075e..f908b59 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java
@@ -19,10 +19,58 @@ package org.apache.ignite.internal.managers.discovery;
import java.io.Serializable;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddedMessage;
import org.jetbrains.annotations.Nullable;
/**
+ * <b>DiscoveryCustomMessage</b> messages are handled by discovery protocol which provides some guarantees around them.
*
+ * When some node sends <b>DiscoveryCustomMessage</b> with {@link GridDiscoveryManager#sendCustomEvent(DiscoveryCustomMessage)}
+ * call, message firstly goes to current coordinator, is verified there and after that gets sent to the cluster.
+ * Only after verification it is delivered to listeners on all nodes starting from coordinator.
+ *
+ * To register a listener {@link GridDiscoveryManager#setCustomEventListener(Class, CustomEventListener)} method is used.
+ *
+ * Discovery protocol guarantees include:
+ * <ol>
+ * <li>
+ * All discovery messages are observed by all nodes in exactly the same order,
+ * it is guaranteed by handling them in single-threaded mode.
+ * </li>
+ * <li>
+ * New server node joining process in default implementation involves two passes of different messages across the cluster:
+ * {@link TcpDiscoveryNodeAddedMessage} and
+ * {@link TcpDiscoveryNodeAddFinishedMessage} messages.
+ * It is guaranteed that all discovery messages observed by coordinator in between these two messages
+ * are reordered and guaranteed to be delivered to newly joined node.
+ * </li>
+ * </ol>
+ *
+ * Yet there are some features and limitations one should be aware of when using custom discovery messaging mechanism:
+ * <ol>
+ * <li>
+ * Guarantee #2 doesn't encompass <b>DiscoveryCustomMessage</b>s created automatically on
+ * {@link DiscoveryCustomMessage#ackMessage()} method call.
+ *
+ * If there were messages of this type in between <b>TcpDiscoveryNodeAddedMessage</b> and
+ * <b>TcpDiscoveryNodeAddFinishedMessage</b> messages, they won't be delivered to new joiner node.
+ * </li>
+ * <li>
+ * There is no guarantee for a given <b>DiscoveryCustomMessage</b> to be delivered only once.
+ * It is possible that because of node failure antecedent node will resend messages
+ * it thinks were not sent by failed node.
+ * Duplicate messages are not filtered out on receiver side.
+ * </li>
+ * <li>
+ * <b>DiscoveryCustomMessage</b>s are delivered to client nodes in asynchronous fashion
+ * as clients don't participate in the cluster ring.
+ * </li>
+ * <li>
+ * Any blocking operations like obtaining locks or doing I/O <b>must</b> be avoided in message handlers
+ * as they may lead to deadlocks and cluster failures.
+ * </li>
+ * </ol>
*/
public interface DiscoveryCustomMessage extends Serializable {
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
index 9194b10..77bd7d4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
@@ -42,7 +42,6 @@ import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteDeploymentCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.GridManagerAdapter;
import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
@@ -71,6 +70,7 @@ import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.events.EventType.EVT_NODE_METRICS_UPDATED;
import static org.apache.ignite.internal.GridTopic.TOPIC_EVENT;
+import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL;
/**
@@ -494,7 +494,7 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
* @return {@code true} if this is a system hidden event.
*/
private boolean isHiddenEvent(int type) {
- return type == EVT_NODE_METRICS_UPDATED;
+ return type == EVT_NODE_METRICS_UPDATED || type == EVT_DISCOVERY_CUSTOM_EVT;
}
/**
@@ -507,7 +507,7 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
* @return {@code true} if this is an internal event.
*/
private boolean isInternalEvent(int type) {
- return type == DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT || F.contains(EVTS_DISCOVERY_ALL, type);
+ return type == EVT_DISCOVERY_CUSTOM_EVT || F.contains(EVTS_DISCOVERY_ALL, type);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataHolder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataHolder.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataHolder.java
new file mode 100644
index 0000000..bbc929e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataHolder.java
@@ -0,0 +1,73 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.io.Serializable;
+import org.apache.ignite.internal.binary.BinaryMetadata;
+
+/**
+ * Wrapper for {@link BinaryMetadata} which is stored in metadata local cache on each node.
+ * Used internally to track version counters (see javadoc for {@link MetadataUpdateProposedMessage} for more details).
+ */
+final class BinaryMetadataHolder implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final BinaryMetadata metadata;
+
+ /** */
+ private final int pendingVer;
+
+ /** */
+ private final int acceptedVer;
+
+
+ /**
+ * @param metadata Metadata.
+ * @param pendingVer Version of this metadata - how many updates were issued for this type.
+ * @param acceptedVer Pending updates count.
+ */
+ BinaryMetadataHolder(BinaryMetadata metadata, int pendingVer, int acceptedVer) {
+ assert metadata != null;
+
+ this.metadata = metadata;
+ this.pendingVer = pendingVer;
+ this.acceptedVer = acceptedVer;
+ }
+
+ /**
+ *
+ */
+ BinaryMetadata metadata() {
+ return metadata;
+ }
+
+ /**
+ *
+ */
+ int pendingVersion() {
+ return pendingVer;
+ }
+
+ /**
+ *
+ */
+ int acceptedVersion() {
+ return acceptedVer;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java
new file mode 100644
index 0000000..770fa32
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java
@@ -0,0 +1,641 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.util.List;
+import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.events.DiscoveryEvent;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.binary.BinaryMetadata;
+import org.apache.ignite.internal.binary.BinaryUtils;
+import org.apache.ignite.internal.managers.communication.GridIoManager;
+import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.discovery.CustomEventListener;
+import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
+import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.jetbrains.annotations.Nullable;
+import org.jsr166.ConcurrentHashMap8;
+
+import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
+import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
+import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYSTEM_POOL;
+
+/**
+ * Provides API for discovery-based metadata exchange protocol and communication SPI-based metadata request protocol.
+ *
+ * It is responsible for sending update and metadata requests and manages message listeners for them.
+ *
+ * It also manages synchronization logic (blocking/unblocking threads requesting updates or up-to-date metadata etc)
+ * around protocols.
+ */
+final class BinaryMetadataTransport {
+ /** */
+ private final GridDiscoveryManager discoMgr;
+
+ /** */
+ private final GridKernalContext ctx;
+
+ /** */
+ private final IgniteLogger log;
+
+ /** */
+ private final UUID locNodeId;
+
+ /** */
+ private final boolean clientNode;
+
+ /** */
+ private final ConcurrentMap<Integer, BinaryMetadataHolder> metaLocCache;
+
+ /** */
+ private final Queue<MetadataUpdateResultFuture> unlabeledFutures = new ConcurrentLinkedQueue<>();
+
+ /** */
+ private final ConcurrentMap<SyncKey, MetadataUpdateResultFuture> syncMap = new ConcurrentHashMap8<>();
+
+ /** */
+ private final ConcurrentMap<Integer, ClientMetadataRequestFuture> clientReqSyncMap = new ConcurrentHashMap8<>();
+
+ /** */
+ private volatile boolean stopping;
+
+ /** */
+ private final List<BinaryMetadataUpdatedListener> binaryUpdatedLsnrs = new CopyOnWriteArrayList<>();
+
+ /**
+ * @param metaLocCache Metadata locale cache.
+ * @param ctx Context.
+ * @param log Logger.
+ */
+ BinaryMetadataTransport(ConcurrentMap<Integer, BinaryMetadataHolder> metaLocCache, final GridKernalContext ctx, IgniteLogger log) {
+ this.metaLocCache = metaLocCache;
+
+ this.ctx = ctx;
+
+ this.log = log;
+
+ discoMgr = ctx.discovery();
+
+ locNodeId = ctx.localNodeId();
+
+ clientNode = ctx.clientNode();
+
+ discoMgr.setCustomEventListener(MetadataUpdateProposedMessage.class, new MetadataUpdateProposedListener());
+
+ discoMgr.setCustomEventListener(MetadataUpdateAcceptedMessage.class, new MetadataUpdateAcceptedListener());
+
+ GridIoManager ioMgr = ctx.io();
+
+ if (clientNode)
+ ioMgr.addMessageListener(GridTopic.TOPIC_METADATA_REQ, new MetadataResponseListener());
+ else
+ ioMgr.addMessageListener(GridTopic.TOPIC_METADATA_REQ, new MetadataRequestListener(ioMgr));
+
+ if (clientNode)
+ ctx.event().addLocalEventListener(new GridLocalEventListener() {
+ @Override public void onEvent(Event evt) {
+ DiscoveryEvent evt0 = (DiscoveryEvent) evt;
+
+ if (!ctx.isStopping()) {
+ for (ClientMetadataRequestFuture fut : clientReqSyncMap.values())
+ fut.onNodeLeft(evt0.eventNode().id());
+ }
+ }
+ }, EVT_NODE_LEFT, EVT_NODE_FAILED);
+ }
+
+ /**
+ * Adds BinaryMetadata updates {@link BinaryMetadataUpdatedListener listener} to transport.
+ *
+ * @param lsnr Listener.
+ */
+ void addBinaryMetadataUpdateListener(BinaryMetadataUpdatedListener lsnr) {
+ binaryUpdatedLsnrs.add(lsnr);
+ }
+
+ /**
+ * Sends request to cluster proposing update for given metadata.
+ *
+ * @param metadata Metadata proposed for update.
+ * @return Future to wait for update result on.
+ */
+ GridFutureAdapter<MetadataUpdateResult> requestMetadataUpdate(BinaryMetadata metadata) throws IgniteCheckedException {
+ MetadataUpdateResultFuture resFut = new MetadataUpdateResultFuture();
+
+ synchronized (this) {
+ unlabeledFutures.add(resFut);
+
+ if (!stopping)
+ discoMgr.sendCustomEvent(new MetadataUpdateProposedMessage(metadata, locNodeId));
+ else
+ resFut.onDone(MetadataUpdateResult.createUpdateDisabledResult());
+ }
+
+ return resFut;
+ }
+
+ /**
+ * Allows thread to wait for a metadata of given typeId and version to be accepted by the cluster.
+ *
+ * @param typeId ID of binary type.
+ * @param ver version of given binary type (see {@link MetadataUpdateProposedMessage} javadoc for more details).
+ * @return future to wait for update result on.
+ */
+ GridFutureAdapter<MetadataUpdateResult> awaitMetadataUpdate(int typeId, int ver) {
+ SyncKey key = new SyncKey(typeId, ver);
+ MetadataUpdateResultFuture resFut = new MetadataUpdateResultFuture(key);
+
+ MetadataUpdateResultFuture oldFut = syncMap.putIfAbsent(key, resFut);
+
+ if (oldFut != null)
+ resFut = oldFut;
+
+ BinaryMetadataHolder holder = metaLocCache.get(typeId);
+
+ if (holder.acceptedVersion() >= ver)
+ resFut.onDone(MetadataUpdateResult.createSuccessfulResult());
+
+ return resFut;
+ }
+
+ /**
+ * Allows client node to request latest version of binary metadata for a given typeId from the cluster
+ * in case client is able to detect that it has obsolete metadata in its local cache.
+ *
+ * @param typeId ID of binary type.
+ * @return future to wait for request arrival on.
+ */
+ GridFutureAdapter<MetadataUpdateResult> requestUpToDateMetadata(int typeId) {
+ ClientMetadataRequestFuture newFut = new ClientMetadataRequestFuture(ctx, typeId, clientReqSyncMap);
+
+ ClientMetadataRequestFuture oldFut = clientReqSyncMap.putIfAbsent(typeId, newFut);
+
+ if (oldFut != null)
+ return oldFut;
+
+ newFut.requestMetadata();
+
+ return newFut;
+ }
+
+ /** */
+ void stop() {
+ stopping = true;
+
+ cancelFutures(MetadataUpdateResult.createUpdateDisabledResult());
+ }
+
+ /** */
+ void onDisconnected() {
+ cancelFutures(MetadataUpdateResult.createFailureResult(new BinaryObjectException("Failed to update or wait for metadata, client node disconnected")));
+ }
+
+ /**
+ * @param res result to cancel futures with.
+ */
+ private void cancelFutures(MetadataUpdateResult res) {
+ for (MetadataUpdateResultFuture fut : unlabeledFutures)
+ fut.onDone(res);
+
+ for (MetadataUpdateResultFuture fut : syncMap.values())
+ fut.onDone(res);
+
+ for (ClientMetadataRequestFuture fut : clientReqSyncMap.values())
+ fut.onDone(res);
+ }
+
+ /** */
+ private final class MetadataUpdateProposedListener implements CustomEventListener<MetadataUpdateProposedMessage> {
+
+ /** {@inheritDoc} */
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, MetadataUpdateProposedMessage msg) {
+ int typeId = msg.typeId();
+
+ BinaryMetadataHolder holder = metaLocCache.get(typeId);
+
+ int pendingVer;
+ int acceptedVer;
+
+ if (msg.pendingVersion() == 0) {
+ if (holder != null) {
+ pendingVer = holder.pendingVersion() + 1;
+ acceptedVer = holder.acceptedVersion();
+ }
+ else {
+ pendingVer = 1;
+ acceptedVer = 0;
+ }
+
+ msg.pendingVersion(pendingVer);
+ msg.acceptedVersion(acceptedVer);
+
+ BinaryMetadata locMeta = holder != null ? holder.metadata() : null;
+
+ try {
+ BinaryMetadata mergedMeta = BinaryUtils.mergeMetadata(locMeta, msg.metadata());
+
+ msg.metadata(mergedMeta);
+ }
+ catch (BinaryObjectException err) {
+ log.warning("Exception with merging metadata for typeId: " + typeId, err);
+
+ msg.markRejected(err);
+ }
+ }
+ else {
+ pendingVer = msg.pendingVersion();
+ acceptedVer = msg.acceptedVersion();
+ }
+
+ if (locNodeId.equals(msg.origNodeId())) {
+ MetadataUpdateResultFuture fut = unlabeledFutures.poll();
+
+ if (msg.rejected())
+ fut.onDone(MetadataUpdateResult.createFailureResult(msg.rejectionError()));
+ else {
+ if (clientNode) {
+ BinaryMetadataHolder newHolder = new BinaryMetadataHolder(msg.metadata(), pendingVer, acceptedVer);
+
+ holder = metaLocCache.putIfAbsent(typeId, newHolder);
+
+ if (holder != null) {
+ boolean obsoleteUpd = false;
+
+ do {
+ holder = metaLocCache.get(typeId);
+
+ if (obsoleteUpdate(
+ holder.pendingVersion(),
+ holder.acceptedVersion(),
+ pendingVer,
+ acceptedVer)) {
+ obsoleteUpd = true;
+
+ fut.onDone(MetadataUpdateResult.createSuccessfulResult());
+
+ break;
+ }
+ }
+ while (!metaLocCache.replace(typeId, holder, newHolder));
+
+ if (!obsoleteUpd)
+ initSyncFor(typeId, pendingVer, fut);
+ }
+ else
+ initSyncFor(typeId, pendingVer, fut);
+ }
+ else {
+ initSyncFor(typeId, pendingVer, fut);
+
+ metaLocCache.put(typeId, new BinaryMetadataHolder(msg.metadata(), pendingVer, acceptedVer));
+ }
+ }
+ }
+ else {
+ if (!msg.rejected()) {
+ BinaryMetadata locMeta = holder != null ? holder.metadata() : null;
+
+ try {
+ BinaryMetadata mergedMeta = BinaryUtils.mergeMetadata(locMeta, msg.metadata());
+
+ BinaryMetadataHolder newHolder = new BinaryMetadataHolder(mergedMeta, pendingVer, acceptedVer);
+
+ if (clientNode) {
+ holder = metaLocCache.putIfAbsent(typeId, newHolder);
+
+ if (holder != null) {
+ do {
+ holder = metaLocCache.get(typeId);
+
+ if (obsoleteUpdate(
+ holder.pendingVersion(),
+ holder.acceptedVersion(),
+ pendingVer,
+ acceptedVer))
+ break;
+
+ } while (!metaLocCache.replace(typeId, holder, newHolder));
+ }
+ }
+ else
+ metaLocCache.put(typeId, newHolder);
+ }
+ catch (BinaryObjectException ignored) {
+ assert false : msg;
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * @param typeId Type ID.
+ * @param pendingVer Pending version.
+ * @param fut Future.
+ */
+ private void initSyncFor(int typeId, int pendingVer, MetadataUpdateResultFuture fut) {
+ if (stopping) {
+ fut.onDone(MetadataUpdateResult.createUpdateDisabledResult());
+
+ return;
+ }
+
+ SyncKey key = new SyncKey(typeId, pendingVer);
+
+ syncMap.put(key, fut);
+
+ fut.key(key);
+ }
+
+ /**
+ *
+ */
+ private final class MetadataUpdateAcceptedListener implements CustomEventListener<MetadataUpdateAcceptedMessage> {
+
+ /** {@inheritDoc} */
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, MetadataUpdateAcceptedMessage msg) {
+ if (msg.duplicated())
+ return;
+
+ int typeId = msg.typeId();
+
+ BinaryMetadataHolder holder = metaLocCache.get(typeId);
+
+ assert holder != null : "No metadata found for typeId " + typeId;
+
+ int newAcceptedVer = msg.acceptedVersion();
+
+ if (clientNode) {
+ BinaryMetadataHolder newHolder = new BinaryMetadataHolder(holder.metadata(),
+ holder.pendingVersion(), newAcceptedVer);
+
+ do {
+ holder = metaLocCache.get(typeId);
+
+ int oldAcceptedVer = holder.acceptedVersion();
+
+ if (oldAcceptedVer > newAcceptedVer)
+ break;
+ }
+ while (!metaLocCache.replace(typeId, holder, newHolder));
+ }
+ else {
+ int oldAcceptedVer = holder.acceptedVersion();
+
+ if (oldAcceptedVer >= newAcceptedVer) {
+ //this is duplicate ack
+ msg.duplicated(true);
+
+ return;
+ }
+
+ metaLocCache.put(typeId, new BinaryMetadataHolder(holder.metadata(), holder.pendingVersion(), newAcceptedVer));
+ }
+
+ for (BinaryMetadataUpdatedListener lsnr : binaryUpdatedLsnrs)
+ lsnr.binaryMetadataUpdated(holder.metadata());
+
+ GridFutureAdapter<MetadataUpdateResult> fut = syncMap.get(new SyncKey(typeId, newAcceptedVer));
+
+ if (fut != null)
+ fut.onDone(MetadataUpdateResult.createSuccessfulResult());
+ }
+ }
+
+ /**
+ * Future class responsible for blocking threads until particular events with metadata updates happen,
+ * e.g. arriving {@link MetadataUpdateAcceptedMessage} acknowledgment or {@link MetadataResponseMessage} response.
+ */
+ private final class MetadataUpdateResultFuture extends GridFutureAdapter<MetadataUpdateResult> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ MetadataUpdateResultFuture() {
+ // No-op.
+ }
+
+ /**
+ * @param key key in syncMap this future was added under.
+ */
+ MetadataUpdateResultFuture(SyncKey key) {
+ this.key = key;
+ }
+
+ /** */
+ private SyncKey key;
+
+ /** {@inheritDoc} */
+ @Override public boolean onDone(@Nullable MetadataUpdateResult res, @Nullable Throwable err) {
+ assert res != null;
+
+ boolean done = super.onDone(res, err);
+
+ if (done && key != null)
+ syncMap.remove(key, this);
+
+ return done;
+ }
+
+ /**
+ * @param key Key.
+ */
+ void key(SyncKey key) {
+ this.key = key;
+ }
+ }
+
+ /**
+ * Key for mapping arriving {@link MetadataUpdateAcceptedMessage} messages
+ * to {@link MetadataUpdateResultFuture}s other threads may be waiting on.
+ */
+ private static final class SyncKey {
+ /** */
+ private final int typeId;
+
+ /** */
+ private final int ver;
+
+ /**
+ * @param typeId Type id.
+ * @param ver Version.
+ */
+ private SyncKey(int typeId, int ver) {
+ this.typeId = typeId;
+ this.ver = ver;
+ }
+
+ /** */
+ int typeId() {
+ return typeId;
+ }
+
+ /** */
+ int version() {
+ return ver;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return typeId + ver;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (o == this)
+ return true;
+
+ if (!(o instanceof SyncKey))
+ return false;
+
+ SyncKey that = (SyncKey) o;
+
+ return (typeId == that.typeId) && (ver == that.ver);
+ }
+ }
+
+ /**
+ * Listener is registered on each server node in cluster waiting for metadata requests from clients.
+ */
+ private final class MetadataRequestListener implements GridMessageListener {
+ /** */
+ private final GridIoManager ioMgr;
+
+ /**
+ * @param ioMgr IO manager.
+ */
+ MetadataRequestListener(GridIoManager ioMgr) {
+ this.ioMgr = ioMgr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ assert msg instanceof MetadataRequestMessage : msg;
+
+ MetadataRequestMessage msg0 = (MetadataRequestMessage) msg;
+
+ int typeId = msg0.typeId();
+
+ BinaryMetadataHolder metaHolder = metaLocCache.get(typeId);
+
+ MetadataResponseMessage resp = new MetadataResponseMessage(typeId);
+
+ byte[] binMetaBytes = null;
+
+ if (metaHolder != null) {
+ try {
+ binMetaBytes = U.marshal(ctx, metaHolder);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to marshal binary metadata for [typeId: " + typeId + "]", e);
+
+ resp.markErrorOnRequest();
+ }
+ }
+
+ resp.binaryMetadataBytes(binMetaBytes);
+
+ try {
+ ioMgr.sendToGridTopic(nodeId, GridTopic.TOPIC_METADATA_REQ, resp, SYSTEM_POOL);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send up-to-date metadata response.", e);
+ }
+ }
+ }
+
+ /**
+ * Listener is registered on each client node and listens for metadata responses from cluster.
+ */
+ private final class MetadataResponseListener implements GridMessageListener {
+
+ /** {@inheritDoc} */
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ assert msg instanceof MetadataResponseMessage : msg;
+
+ MetadataResponseMessage msg0 = (MetadataResponseMessage) msg;
+
+ int typeId = msg0.typeId();
+
+ byte[] binMetaBytes = msg0.binaryMetadataBytes();
+
+ ClientMetadataRequestFuture fut = clientReqSyncMap.get(typeId);
+
+ if (fut == null)
+ return;
+
+ if (msg0.metadataNotFound()) {
+ fut.onDone(MetadataUpdateResult.createSuccessfulResult());
+
+ return;
+ }
+
+ try {
+ BinaryMetadataHolder newHolder = U.unmarshal(ctx, binMetaBytes, U.resolveClassLoader(ctx.config()));
+
+ BinaryMetadataHolder oldHolder = metaLocCache.putIfAbsent(typeId, newHolder);
+
+ if (oldHolder != null) {
+ do {
+ oldHolder = metaLocCache.get(typeId);
+
+ if (oldHolder != null && obsoleteUpdate(
+ oldHolder.pendingVersion(),
+ oldHolder.acceptedVersion(),
+ newHolder.pendingVersion(),
+ newHolder.acceptedVersion()))
+ break;
+ }
+ while (!metaLocCache.replace(typeId, oldHolder, newHolder));
+ }
+
+ fut.onDone(MetadataUpdateResult.createSuccessfulResult());
+ }
+ catch (IgniteCheckedException e) {
+ fut.onDone(MetadataUpdateResult.createFailureResult(new BinaryObjectException(e)));
+ }
+ }
+
+
+ }
+
+ /**
+ * Method checks if arrived metadata is obsolete comparing to the one from local cache.
+ *
+ * @param locP pendingVersion of metadata from local cache.
+ * @param locA acceptedVersion of metadata from local cache.
+ * @param remP pendingVersion of metadata from arrived message (client response/proposed/accepted).
+ * @param remA acceptedVersion of metadata from arrived message (client response/proposed/accepted).
+ * @return {@code true} is
+ */
+ private static boolean obsoleteUpdate(int locP, int locA, int remP, int remA) {
+ return (remP < locP) || (remP == locP && remA < locA);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatedListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatedListener.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatedListener.java
new file mode 100644
index 0000000..08e879a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataUpdatedListener.java
@@ -0,0 +1,29 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import org.apache.ignite.internal.binary.BinaryMetadata;
+
+/**
+ * Interface allows any component to register for events of binary metadata updates.
+ */
+public interface BinaryMetadataUpdatedListener {
+ /**
+ * @param metadata Updated metadata.
+ */
+ public void binaryMetadataUpdated(BinaryMetadata metadata);
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessor.java
index 7cc7a29..1564ff3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessor.java
@@ -71,6 +71,15 @@ public interface CacheObjectBinaryProcessor extends IgniteCacheObjectProcessor {
*/
@Nullable public BinaryType metadata(int typeId) throws IgniteException;
+
+ /**
+ * @param typeId Type ID.
+ * @param schemaId Schema ID.
+ * @return Meta data.
+ * @throws IgniteException In case of error.
+ */
+ @Nullable public BinaryType metadata(int typeId, int schemaId) throws IgniteException;
+
/**
* @param typeIds Type ID.
* @return Meta data.
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
index 6e5940f..50a9f44 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
@@ -17,25 +17,15 @@
package org.apache.ignite.internal.processors.cache.binary;
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import javax.cache.Cache;
+import java.util.concurrent.ConcurrentMap;
import javax.cache.CacheException;
-import javax.cache.event.CacheEntryEvent;
-import javax.cache.event.CacheEntryListenerException;
-import javax.cache.event.CacheEntryUpdatedListener;
-import javax.cache.event.EventType;
-import javax.cache.processor.EntryProcessor;
-import javax.cache.processor.MutableEntry;
import org.apache.ignite.IgniteBinary;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
@@ -45,9 +35,7 @@ import org.apache.ignite.binary.BinaryObjectBuilder;
import org.apache.ignite.binary.BinaryObjectException;
import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.binary.BinaryTypeConfiguration;
-import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.cluster.ClusterTopologyException;
import org.apache.ignite.configuration.BinaryConfiguration;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.events.Event;
@@ -68,45 +56,37 @@ import org.apache.ignite.internal.binary.GridBinaryMarshaller;
import org.apache.ignite.internal.binary.builder.BinaryObjectBuilderImpl;
import org.apache.ignite.internal.binary.streams.BinaryInputStream;
import org.apache.ignite.internal.binary.streams.BinaryOffheapInputStream;
-import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
-import org.apache.ignite.internal.processors.cache.CacheEntryPredicateAdapter;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheObjectContext;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheUtils;
-import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.query.CacheQuery;
-import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessorImpl;
import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.IgniteUtils;
-import org.apache.ignite.internal.util.lang.GridCloseableIterator;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.GridMapEntry;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T1;
import org.apache.ignite.internal.util.typedef.T2;
-import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.spi.IgniteNodeValidationResult;
+import org.apache.ignite.spi.discovery.DiscoveryDataBag;
+import org.apache.ignite.spi.discovery.DiscoveryDataBag.GridDiscoveryData;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SKIP_CONFIGURATION_CONSISTENCY_CHECK;
import static org.apache.ignite.IgniteSystemProperties.getBoolean;
import static org.apache.ignite.events.EventType.EVT_CLIENT_NODE_DISCONNECTED;
+import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.BINARY_PROC;
/**
* Binary processor implementation.
@@ -114,25 +94,7 @@ import static org.apache.ignite.events.EventType.EVT_CLIENT_NODE_DISCONNECTED;
public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorImpl implements
CacheObjectBinaryProcessor {
/** */
- private final CountDownLatch startLatch = new CountDownLatch(1);
-
- /** */
- private final boolean clientNode;
-
- /** */
- private volatile IgniteCacheProxy<BinaryMetadataKey, BinaryMetadata> metaDataCache;
-
- /** */
- private final ConcurrentHashMap8<Integer, BinaryTypeImpl> clientMetaDataCache;
-
- /** Predicate to filter binary meta data in utility cache. */
- private final CacheEntryPredicate metaPred = new CacheEntryPredicateAdapter() {
- private static final long serialVersionUID = 0L;
-
- @Override public boolean apply(GridCacheEntryEx e) {
- return e.key().value(e.context().cacheObjectContext(), false) instanceof BinaryMetadataKey;
- }
- };
+ private volatile boolean discoveryStarted;
/** */
private BinaryContext binaryCtx;
@@ -151,11 +113,16 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
private final GridLocalEventListener clientDisconLsnr = new GridLocalEventListener() {
@Override public void onEvent(Event evt) {
binaryContext().unregisterBinarySchemas();
+
+ metadataLocCache.clear();
}
};
- /** Metadata updates collected before metadata cache is initialized. */
- private final Map<Integer, BinaryMetadata> metaBuf = new ConcurrentHashMap<>();
+ /** Locally cached metadata. This local cache is managed by exchanging discovery custom events. */
+ private final ConcurrentMap<Integer, BinaryMetadataHolder> metadataLocCache = new ConcurrentHashMap8<>();
+
+ /** */
+ private BinaryMetadataTransport transport;
/** Cached affinity key field names. */
private final ConcurrentHashMap<Integer, T1<BinaryField>> affKeyFields = new ConcurrentHashMap<>();
@@ -167,10 +134,6 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
super(ctx);
marsh = ctx.grid().configuration().getMarshaller();
-
- clientNode = this.ctx.clientNode();
-
- clientMetaDataCache = clientNode ? new ConcurrentHashMap8<Integer, BinaryTypeImpl>() : null;
}
/** {@inheritDoc} */
@@ -179,47 +142,38 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
if (ctx.clientNode())
ctx.event().addLocalEventListener(clientDisconLsnr, EVT_CLIENT_NODE_DISCONNECTED);
+ transport = new BinaryMetadataTransport(metadataLocCache, ctx, log);
+
BinaryMetadataHandler metaHnd = new BinaryMetadataHandler() {
@Override public void addMeta(int typeId, BinaryType newMeta) throws BinaryObjectException {
assert newMeta != null;
assert newMeta instanceof BinaryTypeImpl;
- BinaryMetadata newMeta0 = ((BinaryTypeImpl)newMeta).metadata();
+ if (!discoveryStarted) {
+ BinaryMetadataHolder holder = metadataLocCache.get(typeId);
- if (metaDataCache == null) {
- BinaryMetadata oldMeta = metaBuf.get(typeId);
- BinaryMetadata mergedMeta = BinaryUtils.mergeMetadata(oldMeta, newMeta0);
+ BinaryMetadata oldMeta = holder != null ? holder.metadata() : null;
- if (oldMeta != mergedMeta) {
- synchronized (this) {
- mergedMeta = BinaryUtils.mergeMetadata(oldMeta, newMeta0);
+ BinaryMetadata mergedMeta = BinaryUtils.mergeMetadata(oldMeta, ((BinaryTypeImpl)newMeta).metadata());
- if (oldMeta != mergedMeta)
- metaBuf.put(typeId, mergedMeta);
- else
- return;
- }
+ if (oldMeta != mergedMeta)
+ metadataLocCache.putIfAbsent(typeId, new BinaryMetadataHolder(mergedMeta, 0, 0));
- if (metaDataCache == null)
- return;
- else
- metaBuf.remove(typeId);
- }
- else
- return;
+ return;
}
- assert metaDataCache != null;
+ BinaryMetadata newMeta0 = ((BinaryTypeImpl)newMeta).metadata();
CacheObjectBinaryProcessorImpl.this.addMeta(typeId, newMeta0.wrap(binaryCtx));
}
@Override public BinaryType metadata(int typeId) throws BinaryObjectException {
- if (metaDataCache == null)
- U.awaitQuiet(startLatch);
-
return CacheObjectBinaryProcessorImpl.this.metadata(typeId);
}
+
+ @Override public BinaryType metadata(int typeId, int schemaId) throws BinaryObjectException {
+ return CacheObjectBinaryProcessorImpl.this.metadata(typeId, schemaId);
+ }
};
BinaryMarshaller bMarsh0 = (BinaryMarshaller)marsh;
@@ -265,111 +219,34 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
}
}
+ /**
+ * @param lsnr Listener.
+ */
+ public void addBinaryMetadataUpdateListener(BinaryMetadataUpdatedListener lsnr) {
+ if (transport != null)
+ transport.addBinaryMetadataUpdateListener(lsnr);
+ }
+
/** {@inheritDoc} */
@Override public void stop(boolean cancel) {
if (ctx.clientNode())
ctx.event().removeLocalEventListener(clientDisconLsnr);
- }
- /** {@inheritDoc} */
- @Override public void onContinuousProcessorStarted(GridKernalContext ctx) throws IgniteCheckedException {
- if (clientNode && !ctx.isDaemon()) {
- ctx.continuous().registerStaticRoutine(
- CU.UTILITY_CACHE_NAME,
- new MetaDataEntryListener(),
- new MetaDataEntryFilter(),
- null);
- }
+ if (transport != null)
+ transport.stop();
}
/** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override public void onUtilityCacheStarted() throws IgniteCheckedException {
- IgniteCacheProxy<Object, Object> proxy = ctx.cache().jcache(CU.UTILITY_CACHE_NAME);
-
- boolean old = proxy.context().deploy().ignoreOwnership(true);
-
- try {
- metaDataCache = (IgniteCacheProxy)proxy.withNoRetries();
- }
- finally {
- proxy.context().deploy().ignoreOwnership(old);
- }
-
- if (clientNode) {
- assert !metaDataCache.context().affinityNode();
-
- while (true) {
- ClusterNode oldestSrvNode = ctx.discovery().oldestAliveCacheServerNode(AffinityTopologyVersion.NONE);
-
- if (oldestSrvNode == null)
- break;
-
- GridCacheQueryManager qryMgr = metaDataCache.context().queries();
-
- CacheQuery<Map.Entry<BinaryMetadataKey, BinaryMetadata>> qry =
- qryMgr.createScanQuery(new MetaDataPredicate(), null, false);
-
- qry.keepAll(false);
-
- qry.projection(ctx.cluster().get().forNode(oldestSrvNode));
-
- try (GridCloseableIterator<Map.Entry<BinaryMetadataKey, BinaryMetadata>> entries = qry.executeScanQuery()) {
- for (Map.Entry<BinaryMetadataKey, BinaryMetadata> e : entries) {
- assert e.getKey() != null : e;
- assert e.getValue() != null : e;
-
- addClientCacheMetaData(e.getKey(), e.getValue());
- }
- }
- catch (IgniteCheckedException e) {
- if (!ctx.discovery().alive(oldestSrvNode) || !ctx.discovery().pingNode(oldestSrvNode.id()))
- continue;
- else
- throw e;
- }
- catch (CacheException e) {
- if (X.hasCause(e, ClusterTopologyCheckedException.class, ClusterTopologyException.class))
- continue;
- else
- throw e;
- }
-
- break;
- }
- }
-
- for (Map.Entry<Integer, BinaryMetadata> e : metaBuf.entrySet())
- addMeta(e.getKey(), e.getValue().wrap(binaryCtx));
-
- metaBuf.clear();
-
- startLatch.countDown();
+ @Override public void onDisconnected(IgniteFuture<?> reconnectFut) throws IgniteCheckedException {
+ if (transport != null)
+ transport.onDisconnected();
}
- /**
- * @param key Metadata key.
- * @param newMeta Metadata.
- */
- private void addClientCacheMetaData(BinaryMetadataKey key, final BinaryMetadata newMeta) {
- int key0 = key.typeId();
-
- clientMetaDataCache.compute(key0, new ConcurrentHashMap8.BiFun<Integer, BinaryTypeImpl, BinaryTypeImpl>() {
- @Override public BinaryTypeImpl apply(Integer key, BinaryTypeImpl oldMeta) {
- BinaryMetadata res;
-
- BinaryMetadata oldMeta0 = oldMeta != null ? oldMeta.metadata() : null;
-
- try {
- res = BinaryUtils.mergeMetadata(oldMeta0, newMeta);
- }
- catch (BinaryObjectException ignored) {
- res = oldMeta0;
- }
+ /** {@inheritDoc} */
+ @Override public void onKernalStart() throws IgniteCheckedException {
+ super.onKernalStart();
- return res != null ? res.wrap(binaryCtx) : null;
- }
- });
+ discoveryStarted = true;
}
/** {@inheritDoc} */
@@ -383,7 +260,7 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
/**
* @param obj Object.
* @return Bytes.
- * @throws org.apache.ignite.binary.BinaryObjectException If failed.
+ * @throws BinaryObjectException If failed.
*/
public byte[] marshal(@Nullable Object obj) throws BinaryObjectException {
byte[] arr = binaryMarsh.marshal(obj);
@@ -397,7 +274,7 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
* @param ptr Off-heap pointer.
* @param forceHeap If {@code true} creates heap-based object.
* @return Object.
- * @throws org.apache.ignite.binary.BinaryObjectException If failed.
+ * @throws BinaryObjectException If failed.
*/
public Object unmarshal(long ptr, boolean forceHeap) throws BinaryObjectException {
assert ptr > 0 : ptr;
@@ -536,54 +413,94 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
BinaryMetadata newMeta0 = ((BinaryTypeImpl)newMeta).metadata();
- final BinaryMetadataKey key = new BinaryMetadataKey(typeId);
-
try {
- BinaryMetadata oldMeta = metaDataCache.localPeek(key);
- BinaryMetadata mergedMeta = BinaryUtils.mergeMetadata(oldMeta, newMeta0);
+ BinaryMetadataHolder metaHolder = metadataLocCache.get(typeId);
- AffinityTopologyVersion topVer = ctx.cache().context().lockedTopologyVersion(null);
+ BinaryMetadata oldMeta = metaHolder != null ? metaHolder.metadata() : null;
- if (topVer == null)
- topVer = ctx.cache().context().exchange().readyAffinityVersion();
+ BinaryMetadata mergedMeta = BinaryUtils.mergeMetadata(oldMeta, newMeta0);
- BinaryObjectException err = metaDataCache.invoke(topVer, key, new MetadataProcessor(mergedMeta));
+ MetadataUpdateResult res = transport.requestMetadataUpdate(mergedMeta).get();
- if (err != null)
- throw err;
+ assert res != null;
+
+ if (res.rejected())
+ throw res.error();
}
- catch (CacheException e) {
+ catch (IgniteCheckedException e) {
throw new BinaryObjectException("Failed to update meta data for type: " + newMeta.typeName(), e);
}
}
/** {@inheritDoc} */
- @Nullable @Override public BinaryType metadata(final int typeId) throws BinaryObjectException {
- try {
- if (clientNode) {
- BinaryType typeMeta = clientMetaDataCache.get(typeId);
+ @Nullable @Override public BinaryType metadata(final int typeId) {
+ BinaryMetadataHolder holder = metadataLocCache.get(typeId);
+
+ if (holder == null) {
+ if (ctx.clientNode()) {
+ try {
+ transport.requestUpToDateMetadata(typeId).get();
- if (typeMeta != null)
- return typeMeta;
+ holder = metadataLocCache.get(typeId);
+ }
+ catch (IgniteCheckedException ignored) {
+ // No-op.
+ }
+ }
+ }
- BinaryMetadata meta = metaDataCache.getTopologySafe(new BinaryMetadataKey(typeId));
+ if (holder != null) {
+ if (holder.pendingVersion() - holder.acceptedVersion() > 0) {
+ GridFutureAdapter<MetadataUpdateResult> fut = transport.awaitMetadataUpdate(typeId, holder.pendingVersion());
- return meta != null ? meta.wrap(binaryCtx) : null;
+ try {
+ fut.get();
+ }
+ catch (IgniteCheckedException ignored) {
+ // No-op.
+ }
}
- else {
- BinaryMetadataKey key = new BinaryMetadataKey(typeId);
- BinaryMetadata meta = metaDataCache.localPeek(key);
+ return holder.metadata().wrap(binaryCtx);
+ }
+ else
+ return null;
+ }
- if (meta == null && !metaDataCache.context().preloader().syncFuture().isDone())
- meta = metaDataCache.getTopologySafe(key);
+ /** {@inheritDoc} */
+ @Nullable @Override public BinaryType metadata(final int typeId, final int schemaId) {
+ BinaryMetadataHolder holder = metadataLocCache.get(typeId);
+
+ if (ctx.clientNode()) {
+ if (holder == null || (holder != null && !holder.metadata().hasSchema(schemaId))) {
+ try {
+ transport.requestUpToDateMetadata(typeId).get();
- return meta != null ? meta.wrap(binaryCtx) : null;
+ holder = metadataLocCache.get(typeId);
+ }
+ catch (IgniteCheckedException ignored) {
+ // No-op.
+ }
}
}
- catch (CacheException e) {
- throw new BinaryObjectException(e);
+ else {
+ if (holder.pendingVersion() - holder.acceptedVersion() > 0) {
+ GridFutureAdapter<MetadataUpdateResult> fut = transport.awaitMetadataUpdate(
+ typeId,
+ holder.pendingVersion());
+
+ try {
+ fut.get();
+ }
+ catch (IgniteCheckedException ignored) {
+ // No-op.
+ }
+
+ holder = metadataLocCache.get(typeId);
+ }
}
+
+ return holder != null ? holder.metadata().wrap(binaryCtx) : null;
}
/** {@inheritDoc} */
@@ -595,12 +512,10 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
for (Integer typeId : typeIds)
keys.add(new BinaryMetadataKey(typeId));
- Map<BinaryMetadataKey, BinaryMetadata> meta = metaDataCache.getAll(keys);
-
- Map<Integer, BinaryType> res = U.newHashMap(meta.size());
+ Map<Integer, BinaryType> res = U.newHashMap(metadataLocCache.size());
- for (Map.Entry<BinaryMetadataKey, BinaryMetadata> e : meta.entrySet())
- res.put(e.getKey().typeId(), e.getValue().wrap(binaryCtx));
+ for (Map.Entry<Integer, BinaryMetadataHolder> e : metadataLocCache.entrySet())
+ res.put(e.getKey(), e.getValue().metadata().wrap(binaryCtx));
return res;
}
@@ -612,22 +527,11 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override public Collection<BinaryType> metadata() throws BinaryObjectException {
- if (clientNode)
- return F.viewReadOnly(clientMetaDataCache.values(), new IgniteClosure<BinaryTypeImpl, BinaryType>() {
- @Override public BinaryType apply(BinaryTypeImpl meta) {
- return meta;
- }
- });
- else {
- return F.viewReadOnly(metaDataCache.entrySetx(metaPred),
- new C1<Cache.Entry<BinaryMetadataKey, BinaryMetadata>, BinaryType>() {
- private static final long serialVersionUID = 0L;
-
- @Override public BinaryType apply(Cache.Entry<BinaryMetadataKey, BinaryMetadata> e) {
- return e.getValue().wrap(binaryCtx);
- }
- });
- }
+ return F.viewReadOnly(metadataLocCache.values(), new IgniteClosure<BinaryMetadataHolder, BinaryType>() {
+ @Override public BinaryType apply(BinaryMetadataHolder metaHolder) {
+ return metaHolder.metadata().wrap(binaryCtx);
+ }
+ });
}
/** {@inheritDoc} */
@@ -907,127 +811,33 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
return null;
}
- /**
- * Processor responsible for metadata update.
- */
- private static class MetadataProcessor
- implements EntryProcessor<BinaryMetadataKey, BinaryMetadata, BinaryObjectException>, Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** */
- private BinaryMetadata newMeta;
-
- /**
- * For {@link Externalizable}.
- */
- public MetadataProcessor() {
- // No-op.
- }
-
- /**
- * @param newMeta New metadata.
- */
- private MetadataProcessor(BinaryMetadata newMeta) {
- assert newMeta != null;
-
- this.newMeta = newMeta;
- }
-
- /** {@inheritDoc} */
- @Override public BinaryObjectException process(MutableEntry<BinaryMetadataKey, BinaryMetadata> entry,
- Object... args) {
- try {
- BinaryMetadata oldMeta = entry.getValue();
-
- BinaryMetadata mergedMeta = BinaryUtils.mergeMetadata(oldMeta, newMeta);
-
- if (mergedMeta != oldMeta)
- entry.setValue(mergedMeta);
-
- return null;
- }
- catch (BinaryObjectException e) {
- return e;
- }
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(newMeta);
- }
-
- /** {@inheritDoc} */
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- newMeta = (BinaryMetadata)in.readObject();
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(MetadataProcessor.class, this);
- }
- }
-
- /**
- *
- */
- class MetaDataEntryListener implements CacheEntryUpdatedListener<BinaryMetadataKey, BinaryMetadata> {
- /** {@inheritDoc} */
- @Override public void onUpdated(
- Iterable<CacheEntryEvent<? extends BinaryMetadataKey, ? extends BinaryMetadata>> evts)
- throws CacheEntryListenerException {
- for (CacheEntryEvent<? extends BinaryMetadataKey, ? extends BinaryMetadata> evt : evts) {
- assert evt.getEventType() == EventType.CREATED || evt.getEventType() == EventType.UPDATED : evt;
-
- BinaryMetadataKey key = evt.getKey();
-
- final BinaryMetadata newMeta = evt.getValue();
-
- assert newMeta != null : evt;
-
- addClientCacheMetaData(key, newMeta);
- }
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(MetaDataEntryListener.class, this);
- }
+ /** {@inheritDoc} */
+ @Nullable @Override public DiscoveryDataExchangeType discoveryDataType() {
+ return BINARY_PROC;
}
- /**
- *
- */
- static class MetaDataEntryFilter implements CacheEntryEventSerializableFilter<Object, Object> {
- /** */
- private static final long serialVersionUID = 0L;
+ /** {@inheritDoc} */
+ @Override public void collectGridNodeData(DiscoveryDataBag dataBag) {
+ if (!dataBag.commonDataCollectedFor(BINARY_PROC.ordinal())) {
+ Map<Integer, BinaryMetadataHolder> res = U.newHashMap(metadataLocCache.size());
- /** {@inheritDoc} */
- @Override public boolean evaluate(CacheEntryEvent<?, ?> evt) throws CacheEntryListenerException {
- return evt.getKey() instanceof BinaryMetadataKey;
- }
+ for (Map.Entry<Integer,BinaryMetadataHolder> e : metadataLocCache.entrySet())
+ res.put(e.getKey(), e.getValue());
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(MetaDataEntryFilter.class, this);
+ dataBag.addGridCommonData(BINARY_PROC.ordinal(), (Serializable) res);
}
}
- /**
- *
- */
- static class MetaDataPredicate implements IgniteBiPredicate<Object, Object> {
- /** */
- private static final long serialVersionUID = 0L;
+ /** {@inheritDoc} */
+ @Override public void onGridDataReceived(GridDiscoveryData data) {
+ Map<Integer, BinaryMetadataHolder> receivedData = (Map<Integer, BinaryMetadataHolder>) data.commonData();
- /** {@inheritDoc} */
- @Override public boolean apply(Object key, Object val) {
- return key instanceof BinaryMetadataKey;
- }
+ if (receivedData != null) {
+ for (Map.Entry<Integer, BinaryMetadataHolder> e : receivedData.entrySet()) {
+ BinaryMetadataHolder holder = e.getValue();
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(MetaDataPredicate.class, this);
+ metadataLocCache.put(e.getKey(), new BinaryMetadataHolder(holder.metadata(), holder.pendingVersion(), holder.pendingVersion()));
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/44cf1d21/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/ClientMetadataRequestFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/ClientMetadataRequestFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/ClientMetadataRequestFuture.java
new file mode 100644
index 0000000..658e9da
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/ClientMetadataRequestFuture.java
@@ -0,0 +1,161 @@
+/*
+ * 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.internal.processors.cache.binary;
+
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.managers.communication.GridIoManager;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Future is responsible for requesting up-to-date metadata from any server node in cluster
+ * and for blocking thread on client node until response arrives.
+ *
+ * It can cope with situation if node currently requested for the metadata leaves cluster;
+ * in that case future simply re-requests metadata from the next node available in topology.
+ */
+final class ClientMetadataRequestFuture extends GridFutureAdapter<MetadataUpdateResult> {
+ /** */
+ private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+
+ /** */
+ private static IgniteLogger log;
+
+ /** */
+ private final GridIoManager ioMgr;
+
+ /** */
+ private final GridDiscoveryManager discoMgr;
+
+ /** */
+ private final int typeId;
+
+ /** */
+ private final Map<Integer, ClientMetadataRequestFuture> syncMap;
+
+ /** */
+ private final Queue<ClusterNode> aliveSrvNodes;
+
+ /** */
+ private ClusterNode pendingNode;
+
+ /**
+ * @param ctx Context.
+ * @param syncMap Map to store futures for ongoing requests.
+ */
+ ClientMetadataRequestFuture(
+ GridKernalContext ctx,
+ int typeId,
+ Map<Integer, ClientMetadataRequestFuture> syncMap
+ ) {
+ ioMgr = ctx.io();
+ discoMgr = ctx.discovery();
+ aliveSrvNodes = new LinkedList<>(discoMgr.aliveServerNodes());
+
+ this.typeId = typeId;
+ this.syncMap = syncMap;
+
+ if (log == null)
+ log = U.logger(ctx, logRef, ClientMetadataRequestFuture.class);
+ }
+
+ /** */
+ void requestMetadata() {
+ boolean noSrvsInCluster;
+
+ synchronized (this) {
+ while (!aliveSrvNodes.isEmpty()) {
+ ClusterNode srvNode = aliveSrvNodes.poll();
+
+ try {
+ ioMgr.sendToGridTopic(srvNode,
+ GridTopic.TOPIC_METADATA_REQ,
+ new MetadataRequestMessage(typeId),
+ GridIoPolicy.SYSTEM_POOL);
+
+ if (discoMgr.node(srvNode.id()) == null)
+ continue;
+
+ pendingNode = srvNode;
+
+ break;
+ }
+ catch (IgniteCheckedException ignored) {
+ U.warn(log,
+ "Failed to request marshaller mapping from remote node (proceeding with the next one): "
+ + srvNode);
+ }
+ }
+
+ noSrvsInCluster = pendingNode == null;
+ }
+
+ if (noSrvsInCluster)
+ onDone(MetadataUpdateResult.createFailureResult(
+ new BinaryObjectException(
+ "All server nodes have left grid, cannot request metadata [typeId: "
+ + typeId + "]")));
+ }
+
+ /**
+ * If left node is the one latest metadata request was sent to,
+ * request is sent again to the next node in topology.
+ *
+ * @param leftNodeId ID of left node.
+ */
+ void onNodeLeft(UUID leftNodeId) {
+ boolean reqAgain = false;
+
+ synchronized (this) {
+ if (pendingNode != null && pendingNode.id().equals(leftNodeId)) {
+ aliveSrvNodes.remove(pendingNode);
+
+ pendingNode = null;
+
+ reqAgain = true;
+ }
+ }
+
+ if (reqAgain)
+ requestMetadata();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onDone(@Nullable MetadataUpdateResult res, @Nullable Throwable err) {
+ assert res != null;
+
+ boolean done = super.onDone(res, err);
+
+ if (done)
+ syncMap.remove(typeId);
+
+ return done;
+ }
+}
[3/9] ignite git commit: IGNITE-4838 Fixed internal task detection
logic. Added tests.
Posted by ag...@apache.org.
IGNITE-4838 Fixed internal task detection logic. Added tests.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ba68c6ca
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ba68c6ca
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ba68c6ca
Branch: refs/heads/ignite-4003
Commit: ba68c6cae674aa6b2d76035b376fcbdf5c4a2f64
Parents: 8dd88d8
Author: Vasiliy Sisko <vs...@gridgain.com>
Authored: Thu Mar 30 11:08:10 2017 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Thu Mar 30 11:08:10 2017 +0700
----------------------------------------------------------------------
.../processors/task/GridTaskProcessor.java | 9 +++++-
.../internal/GridTaskExecutionSelfTest.java | 34 ++++++++++++++++++++
...xecutionWithoutPeerClassLoadingSelfTest.java | 31 ++++++++++++++++++
.../testsuites/IgniteComputeGridTestSuite.java | 2 ++
4 files changed, 75 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ba68c6ca/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java
index 9a98aeb..a11eabc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java
@@ -608,6 +608,13 @@ public class GridTaskProcessor extends GridProcessorAdapter {
if (subjId == null)
subjId = ctx.localNodeId();
+ boolean internal = false;
+
+ if (dep == null || taskCls == null)
+ assert deployEx != null;
+ else
+ internal = dep.internalTask(task, taskCls);
+
// Creates task session with task name and task version.
GridTaskSessionImpl ses = ctx.session().createTaskSession(
sesId,
@@ -621,7 +628,7 @@ public class GridTaskProcessor extends GridProcessorAdapter {
Collections.<ComputeJobSibling>emptyList(),
Collections.emptyMap(),
fullSup,
- dep != null && dep.internalTask(task, taskCls),
+ internal,
subjId);
ComputeTaskInternalFuture<R> fut = new ComputeTaskInternalFuture<>(ses, ctx);
http://git-wip-us.apache.org/repos/asf/ignite/blob/ba68c6ca/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
index a98c578..e533a70 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionSelfTest.java
@@ -22,8 +22,10 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.GridTestTask;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteDeploymentException;
import org.apache.ignite.compute.ComputeJobContext;
import org.apache.ignite.compute.ComputeTaskFuture;
+import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.lang.IgniteFuture;
@@ -45,6 +47,20 @@ public class GridTaskExecutionSelfTest extends GridCommonAbstractTest {
super(false);
}
+ /** */
+ protected boolean peerClassLoadingEnabled() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setPeerClassLoadingEnabled(peerClassLoadingEnabled());
+
+ return cfg;
+ }
+
/** {@inheritDoc} */
@Override protected void beforeTestsStarted() throws Exception {
startGrid(1);
@@ -136,4 +152,22 @@ public class GridTaskExecutionSelfTest extends GridCommonAbstractTest {
for (IgniteFuture<Object> fut : futs)
fut.get();
}
+
+ /**
+ * Test execution of non-existing task by name IGNITE-4838.
+ *
+ * @throws Exception If failed.
+ */
+ public void testExecuteTaskWithInvalidName() throws Exception {
+ try {
+ ComputeTaskFuture<?> fut = ignite.compute().execute("invalid.task.name", null);
+
+ fut.get();
+
+ assert false : "Should never be reached due to exception thrown.";
+ }
+ catch (IgniteDeploymentException e) {
+ info("Received correct exception: " + e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ba68c6ca/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionWithoutPeerClassLoadingSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionWithoutPeerClassLoadingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionWithoutPeerClassLoadingSelfTest.java
new file mode 100644
index 0000000..45e65cd
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskExecutionWithoutPeerClassLoadingSelfTest.java
@@ -0,0 +1,31 @@
+/*
+ * 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.internal;
+
+import org.apache.ignite.testframework.junits.common.GridCommonTest;
+
+/**
+ * Task execution test.
+ */
+@GridCommonTest(group = "Kernal Self")
+public class GridTaskExecutionWithoutPeerClassLoadingSelfTest extends GridTaskExecutionSelfTest {
+ /** {@inheritDoc} */
+ @Override protected boolean peerClassLoadingEnabled() {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ba68c6ca/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java
index 66238d9..810c488 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java
@@ -52,6 +52,7 @@ import org.apache.ignite.internal.GridProjectionLocalJobMultipleArgumentsSelfTes
import org.apache.ignite.internal.GridStopWithWaitSelfTest;
import org.apache.ignite.internal.GridTaskCancelSingleNodeSelfTest;
import org.apache.ignite.internal.GridTaskExecutionSelfTest;
+import org.apache.ignite.internal.GridTaskExecutionWithoutPeerClassLoadingSelfTest;
import org.apache.ignite.internal.GridTaskFailoverAffinityRunTest;
import org.apache.ignite.internal.GridTaskFailoverSelfTest;
import org.apache.ignite.internal.GridTaskFutureImplStopGridSelfTest;
@@ -102,6 +103,7 @@ public class IgniteComputeGridTestSuite {
suite.addTestSuite(GridCancelUnusedJobSelfTest.class);
suite.addTestSuite(GridTaskJobRejectSelfTest.class);
suite.addTestSuite(GridTaskExecutionSelfTest.class);
+ suite.addTestSuite(GridTaskExecutionWithoutPeerClassLoadingSelfTest.class);
suite.addTestSuite(GridFailoverSelfTest.class);
suite.addTestSuite(GridTaskListenerSelfTest.class);
suite.addTestSuite(GridFailoverTopologySelfTest.class);
[4/9] ignite git commit: .NET: Remove unused imports
Posted by ag...@apache.org.
.NET: Remove unused imports
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f056f2e3
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f056f2e3
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f056f2e3
Branch: refs/heads/ignite-4003
Commit: f056f2e390fb836124e4994add8ca92e267aa754
Parents: ba68c6c
Author: Pavel Tupitsyn <pt...@apache.org>
Authored: Thu Mar 30 09:37:50 2017 +0300
Committer: Pavel Tupitsyn <pt...@apache.org>
Committed: Thu Mar 30 09:37:50 2017 +0300
----------------------------------------------------------------------
.../Apache.Ignite.Core/Impl/Binary/Io/IBinaryStreamProcessor.cs | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f056f2e3/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/IBinaryStreamProcessor.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/IBinaryStreamProcessor.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/IBinaryStreamProcessor.cs
index 1becf80..c76e03d 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/IBinaryStreamProcessor.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/IBinaryStreamProcessor.cs
@@ -17,8 +17,6 @@
namespace Apache.Ignite.Core.Impl.Binary.IO
{
- using System;
-
/// <summary>
/// Binary stream processor.
/// </summary>
[9/9] ignite git commit: ignite-4003 Async outgoing connections for
communication SPI
Posted by ag...@apache.org.
ignite-4003 Async outgoing connections for communication SPI
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/04aca2e6
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/04aca2e6
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/04aca2e6
Branch: refs/heads/ignite-4003
Commit: 04aca2e62f6ac9c50fb0223fdd3d328a98ec81e0
Parents: 44cf1d2
Author: agura <ag...@apache.org>
Authored: Tue Feb 7 14:45:57 2017 +0300
Committer: agura <ag...@apache.org>
Committed: Thu Mar 30 17:31:42 2017 +0300
----------------------------------------------------------------------
.../GridClientConnectionManagerAdapter.java | 1 -
.../connection/GridClientNioTcpConnection.java | 2 +-
.../managers/communication/GridIoManager.java | 4 +
.../internal/util/GridSpinReadWriteLock.java | 2 +-
.../util/nio/GridNioRecoveryDescriptor.java | 2 +-
.../ignite/internal/util/nio/GridNioServer.java | 218 +-
.../util/nio/GridSelectorNioSessionImpl.java | 2 -
.../internal/util/nio/ssl/GridNioSslFilter.java | 12 +-
.../communication/tcp/TcpCommunicationSpi.java | 1902 ++++++++++++------
.../IgniteCacheMessageWriteTimeoutTest.java | 4 +-
.../internal/util/nio/GridNioSelfTest.java | 2 +-
.../spi/GridTcpSpiForwardingSelfTest.java | 3 +-
.../GridAbstractCommunicationSelfTest.java | 27 +-
...mmunicationSpiConcurrentConnectSelfTest.java | 28 +-
...dTcpCommunicationSpiRecoveryAckSelfTest.java | 51 +-
...GridTcpCommunicationSpiRecoverySelfTest.java | 49 +-
...CommunicationRecoveryAckClosureSelfTest.java | 36 +-
.../tcp/TcpCommunicationSpiDropNodesTest.java | 12 +-
.../TcpCommunicationSpiFaultyClientTest.java | 5 +-
.../HadoopExternalCommunication.java | 5 +-
20 files changed, 1659 insertions(+), 708 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientConnectionManagerAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientConnectionManagerAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientConnectionManagerAdapter.java
index e325897..aa06322 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientConnectionManagerAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientConnectionManagerAdapter.java
@@ -183,7 +183,6 @@ public abstract class GridClientConnectionManagerAdapter implements GridClientCo
GridNioSslFilter sslFilter = new GridNioSslFilter(sslCtx, true, ByteOrder.nativeOrder(), gridLog);
sslFilter.directMode(false);
- sslFilter.clientMode(true);
filters = new GridNioFilter[]{codecFilter, sslFilter};
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
index d3a30fb..73487db 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
@@ -234,7 +234,7 @@ public class GridClientNioTcpConnection extends GridClientConnection {
meta.put(GridNioSslFilter.HANDSHAKE_FUT_META_KEY, sslHandshakeFut);
}
- ses = (GridNioSession)srv.createSession(ch, meta).get();
+ ses = (GridNioSession)srv.createSession(ch, meta, false, null).get();
if (sslHandshakeFut != null)
sslHandshakeFut.get();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
index 23738d7..40905f8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
@@ -48,6 +48,7 @@ import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteComponentType;
import org.apache.ignite.internal.IgniteDeploymentCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.direct.DirectMessageReader;
import org.apache.ignite.internal.direct.DirectMessageWriter;
import org.apache.ignite.internal.managers.GridManagerAdapter;
@@ -1298,6 +1299,9 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
((TcpCommunicationSpi)(CommunicationSpi)getSpi()).sendMessage(node, ioMsg, ackC);
else
getSpi().sendMessage(node, ioMsg);
+
+ if (ctx.discovery().node(node.id()) == null)
+ throw new ClusterTopologyCheckedException("Failed to send message to node, node left: " + node);
}
catch (IgniteSpiException e) {
throw new IgniteCheckedException("Failed to send message (node may have left the grid or " +
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinReadWriteLock.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinReadWriteLock.java b/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinReadWriteLock.java
index 4f23979..8fef887 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinReadWriteLock.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinReadWriteLock.java
@@ -70,7 +70,7 @@ public class GridSpinReadWriteLock {
private int writeLockEntryCnt;
/**
- * Acquires write lock.
+ * Acquires read lock.
*/
@SuppressWarnings("BusyWait")
public void readLock() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioRecoveryDescriptor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioRecoveryDescriptor.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioRecoveryDescriptor.java
index 6258c13..af7b757 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioRecoveryDescriptor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioRecoveryDescriptor.java
@@ -259,7 +259,7 @@ public class GridNioRecoveryDescriptor {
/**
* @param node Node.
- * @return {@code True} if node is not null and has the same order as initial remtoe node.
+ * @return {@code True} if node is not null and has the same order as initial remote node.
*/
public boolean nodeAlive(@Nullable ClusterNode node) {
return node != null && node.order() == this.node.order();
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
----------------------------------------------------------------------
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 7f25e40..0848a8b 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
@@ -110,6 +110,12 @@ public class GridNioServer<T> {
/** SSL write buf limit. */
private static final int WRITE_BUF_LIMIT = GridNioSessionMetaKey.nextUniqueKey();
+ /** Session future meta key. */
+ private static final int SESSION_FUT_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
+
+ /** Selection key meta key. */
+ private static final int WORKER_IDX_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
+
/** */
private static final boolean DISABLE_KEYSET_OPTIMIZATION =
IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_NO_SELECTOR_OPTS);
@@ -462,7 +468,7 @@ public class GridNioServer<T> {
* @return Future for operation.
*/
public GridNioFuture<Boolean> close(GridNioSession ses) {
- assert ses instanceof GridSelectorNioSessionImpl;
+ assert ses instanceof GridSelectorNioSessionImpl : ses;
GridSelectorNioSessionImpl impl = (GridSelectorNioSessionImpl)ses;
@@ -706,6 +712,7 @@ public class GridNioServer<T> {
/**
*
*/
+ @SuppressWarnings("ForLoopReplaceableByForEach")
public void dumpStats() {
U.warn(log, "NIO server statistics [readerSesBalanceCnt=" + readerMoveCnt.get() +
", writerSesBalanceCnt=" + writerMoveCnt.get() + ']');
@@ -719,17 +726,34 @@ public class GridNioServer<T> {
*
* @param ch Channel to register within the server and create session for.
* @param meta Optional meta for new session.
+ * @param async Async connection.
+ * @param lsnr Listener that should be invoked in NIO thread.
* @return Future to get session.
*/
- public GridNioFuture<GridNioSession> createSession(final SocketChannel ch,
- @Nullable Map<Integer, ?> meta) {
+ public GridNioFuture<GridNioSession> createSession(
+ final SocketChannel ch,
+ @Nullable Map<Integer, Object> meta,
+ boolean async,
+ @Nullable IgniteInClosure<? super IgniteInternalFuture<GridNioSession>> lsnr
+ ) {
try {
if (!closed) {
ch.configureBlocking(false);
NioOperationFuture<GridNioSession> req = new NioOperationFuture<>(ch, false, meta);
- offerBalanced(req);
+ if (async) {
+ assert meta != null;
+
+ req.op = NioOperation.CONNECT;
+
+ meta.put(SESSION_FUT_META_KEY, req);
+ }
+
+ if (lsnr != null)
+ req.listen(lsnr);
+
+ offerBalanced(req, meta);
return req;
}
@@ -743,6 +767,29 @@ public class GridNioServer<T> {
}
/**
+ * @param ch Channel.
+ * @param meta Session meta.
+ */
+ public GridNioFuture<GridNioSession> cancelConnect(final SocketChannel ch, Map<Integer, ?> meta) {
+ if (!closed) {
+ NioOperationFuture<GridNioSession> req = new NioOperationFuture<>(ch, false, meta);
+
+ req.op = NioOperation.CANCEL_CONNECT;
+
+ Integer idx = (Integer)meta.get(WORKER_IDX_META_KEY);
+
+ assert idx != null : meta;
+
+ clientWorkers.get(idx).offer(req);
+
+ return req;
+ }
+ else
+ return new GridNioFinishedFuture<>(
+ new IgniteCheckedException("Failed to cancel connection, server is stopped."));
+ }
+
+ /**
* Gets configurable write timeout for this session. If not set, default value is {@link #DFLT_SES_WRITE_TIMEOUT}.
*
* @return Write timeout in milliseconds.
@@ -828,9 +875,11 @@ public class GridNioServer<T> {
/**
* @param req Request to balance.
+ * @param meta Session metadata.
+ * @return Worker index.
*/
- private synchronized void offerBalanced(NioOperationFuture req) {
- assert req.operation() == NioOperation.REGISTER : req;
+ private synchronized int offerBalanced(NioOperationFuture req, @Nullable Map<Integer, Object> meta) {
+ assert req.operation() == NioOperation.REGISTER || req.operation() == NioOperation.CONNECT: req;
assert req.socketChannel() != null : req;
int workers = clientWorkers.size();
@@ -868,7 +917,12 @@ public class GridNioServer<T> {
else
balanceIdx = 0;
+ if (meta != null)
+ meta.put(WORKER_IDX_META_KEY, balanceIdx);
+
clientWorkers.get(balanceIdx).offer(req);
+
+ return balanceIdx;
}
/** {@inheritDoc} */
@@ -1692,6 +1746,38 @@ public class GridNioServer<T> {
while ((req0 = changeReqs.poll()) != null) {
switch (req0.operation()) {
+ case CONNECT: {
+ NioOperationFuture req = (NioOperationFuture)req0;
+
+ SocketChannel ch = req.socketChannel();
+
+ try {
+ ch.register(selector, SelectionKey.OP_CONNECT, req.meta());
+ }
+ catch (IOException e) {
+ req.onDone(new IgniteCheckedException("Failed to register channel on selector", e));
+ }
+
+ break;
+ }
+
+ case CANCEL_CONNECT: {
+ NioOperationFuture req = (NioOperationFuture)req0;
+
+ SocketChannel ch = req.socketChannel();
+
+ SelectionKey key = ch.keyFor(selector);
+
+ if (key != null)
+ key.cancel();
+
+ U.closeQuiet(ch);
+
+ req.onDone();
+
+ break;
+ }
+
case REGISTER: {
register((NioOperationFuture)req0);
@@ -1898,8 +1984,12 @@ public class GridNioServer<T> {
log.debug("Closing all connected client sockets.");
// Close all channels registered with selector.
- for (SelectionKey key : selector.keys())
- close((GridSelectorNioSessionImpl)key.attachment(), null);
+ for (SelectionKey key : selector.keys()) {
+ Object attach = key.attachment();
+
+ if (attach instanceof GridSelectorNioSessionImpl)
+ close((GridSelectorNioSessionImpl)attach, null);
+ }
if (log.isDebugEnabled())
log.debug("Closing NIO selector.");
@@ -2022,11 +2112,19 @@ public class GridNioServer<T> {
if (!key.isValid())
continue;
- GridSelectorNioSessionImpl ses = (GridSelectorNioSessionImpl)key.attachment();
-
- assert ses != null;
+ GridSelectorNioSessionImpl ses = null;
try {
+ if (key.isConnectable()) {
+ processConnect(key);
+
+ continue;
+ }
+
+ ses = (GridSelectorNioSessionImpl)key.attachment();
+
+ assert ses != null;
+
if (key.isReadable())
processRead(key);
@@ -2038,9 +2136,11 @@ public class GridNioServer<T> {
throw e;
}
catch (Exception e) {
- U.warn(log, "Failed to process selector key (will close): " + ses, e);
+ if (!closed)
+ U.error(log, "Failed to process selector key [ses=" + ses + ']', e);
- close(ses, new GridNioException(e));
+ if (ses != null)
+ close(ses, new GridNioException(e));
}
}
}
@@ -2067,11 +2167,19 @@ public class GridNioServer<T> {
if (!key.isValid())
continue;
- GridSelectorNioSessionImpl ses = (GridSelectorNioSessionImpl)key.attachment();
-
- assert ses != null;
+ GridSelectorNioSessionImpl ses = null;
try {
+ if (key.isConnectable()) {
+ processConnect(key);
+
+ continue;
+ }
+
+ ses = (GridSelectorNioSessionImpl)key.attachment();
+
+ assert ses != null;
+
if (key.isReadable())
processRead(key);
@@ -2084,9 +2192,10 @@ public class GridNioServer<T> {
}
catch (Exception e) {
if (!closed)
- U.warn(log, "Failed to process selector key (will close): " + ses, e);
+ U.error(log, "Failed to process selector key [ses=" + ses + ']', e);
- close(ses, new GridNioException(e));
+ if (ses != null)
+ close(ses, new GridNioException(e));
}
}
}
@@ -2100,7 +2209,12 @@ public class GridNioServer<T> {
long now = U.currentTimeMillis();
for (SelectionKey key : keys) {
- GridSelectorNioSessionImpl ses = (GridSelectorNioSessionImpl)key.attachment();
+ Object obj = key.attachment();
+
+ if (!(obj instanceof GridSelectorNioSessionImpl))
+ continue;
+
+ GridSelectorNioSessionImpl ses = (GridSelectorNioSessionImpl)obj;
try {
long writeTimeout0 = writeTimeout;
@@ -2181,12 +2295,32 @@ public class GridNioServer<T> {
ses.addMeta(e.getKey(), e.getValue());
}
- SelectionKey key = sockCh.register(selector, SelectionKey.OP_READ, ses);
+ SelectionKey key;
- ses.key(key);
+ if (!sockCh.isRegistered())
+ key = sockCh.register(selector, SelectionKey.OP_READ, ses);
+ else {
+ key = sockCh.keyFor(selector);
+
+ Map<Integer, Object> m = (Map<Integer, Object>)key.attachment();
+
+ NioOperationFuture<GridNioSession> fut =
+ (NioOperationFuture<GridNioSession>)m.remove(SESSION_FUT_META_KEY);
+
+ assert fut != null;
+
+ for (Entry<Integer, Object> e : m.entrySet())
+ ses.addMeta(e.getKey(), e.getValue());
+
+ key.attach(ses);
+
+ key.interestOps(key.interestOps() & (~SelectionKey.OP_CONNECT));
+ key.interestOps(key.interestOps() | SelectionKey.OP_READ);
- if (!ses.accepted())
- resend(ses);
+ fut.onDone(ses);
+ }
+
+ ses.key(key);
sessions.add(ses);
workerSessions.add(ses);
@@ -2321,6 +2455,34 @@ public class GridNioServer<T> {
}
/**
+ * @param key Key.
+ * @throws IOException If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private void processConnect(SelectionKey key) throws IOException {
+ SocketChannel ch = (SocketChannel)key.channel();
+
+ Map<Integer, Object> meta = (Map<Integer, Object>)key.attachment();
+
+ try {
+ if (ch.finishConnect())
+ register(new NioOperationFuture<GridNioSession>(ch, false, meta));
+ }
+ catch (IOException e) {
+ NioOperationFuture<GridNioSession> sesFut =
+ (NioOperationFuture<GridNioSession>)meta.get(SESSION_FUT_META_KEY);
+
+ assert sesFut != null;
+
+ U.closeQuiet(ch);
+
+ sesFut.onDone(new GridNioException("Failed to connect to node", e));
+
+ throw e;
+ }
+ }
+
+ /**
* Processes read-available event on the key.
*
* @param key Key that is ready to be read.
@@ -2537,14 +2699,20 @@ public class GridNioServer<T> {
* @param sockCh Socket channel to be registered on one of the selectors.
*/
private void addRegistrationReq(SocketChannel sockCh) {
- offerBalanced(new NioOperationFuture(sockCh));
+ offerBalanced(new NioOperationFuture(sockCh), null);
}
}
/**
* Asynchronous operation that may be requested on selector.
*/
- enum NioOperation {
+ private enum NioOperation {
+ /** Register connect key selection. */
+ CONNECT,
+
+ /** Cancel connect. */
+ CANCEL_CONNECT,
+
/** Register read key selection. */
REGISTER,
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
index 66f9176..2280321 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
@@ -393,8 +393,6 @@ class GridSelectorNioSessionImpl extends GridNioSessionImpl {
if (!outRecovery.pairedConnections())
inRecovery = outRecovery;
- outRecovery.onConnected();
-
return null;
}
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/04aca2e6/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
index b4bd34a..f8a0dce 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
@@ -69,9 +69,6 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
/** Allocate direct buffer or heap buffer. */
private boolean directBuf;
- /** Whether SSLEngine should use client mode. */
- private boolean clientMode;
-
/** Whether direct mode is used. */
private boolean directMode;
@@ -93,13 +90,6 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
}
/**
- * @param clientMode Flag indicating whether SSLEngine should use client mode..
- */
- public void clientMode(boolean clientMode) {
- this.clientMode = clientMode;
- }
-
- /**
*
* @param directMode Flag indicating whether direct mode is used.
*/
@@ -164,6 +154,8 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
if (sslMeta == null) {
engine = sslCtx.createSSLEngine();
+ boolean clientMode = !ses.accepted();
+
engine.setUseClientMode(clientMode);
if (!clientMode) {