You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2016/10/10 12:13:15 UTC
[19/50] [abbrv] ignite git commit: ignite-3220 Use pair of
connections (in/out) instead of single connection,
possibility to configure mutiple connection pairs.
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/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 c7f7ad4..ed047fa 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
@@ -31,6 +31,7 @@ import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
@@ -78,6 +79,9 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
/** */
private static int port = 60_000;
+ /** */
+ private int connectionsPerNode = 1;
+
/**
*
*/
@@ -158,6 +162,17 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
/**
* @throws Exception If failed.
*/
+ public void testMultithreaded_10Connections() throws Exception {
+ connectionsPerNode = 10;
+
+ int threads = Runtime.getRuntime().availableProcessors() * 5;
+
+ concurrentConnect(threads, 10, 10, false, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testWithLoad() throws Exception {
int threads = Runtime.getRuntime().availableProcessors() * 5;
@@ -239,7 +254,7 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
final AtomicInteger idx = new AtomicInteger();
try {
- GridTestUtils.runMultiThreaded(new Callable<Void>() {
+ final Callable<Void> c = new Callable<Void>() {
@Override public Void call() throws Exception {
int idx0 = idx.getAndIncrement();
@@ -265,7 +280,40 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
return null;
}
- }, threads, "test");
+ };
+
+ List<Thread> threadsList = new ArrayList<>();
+
+ final AtomicBoolean fail = new AtomicBoolean();
+
+ final AtomicLong tId = new AtomicLong();
+
+ for (int t = 0; t < threads; t++) {
+ Thread t0 = new Thread(new Runnable() {
+ @Override public void run() {
+ try {
+ c.call();
+ }
+ catch (Throwable e) {
+ log.error("Unexpected error: " + e, e);
+
+ fail.set(true);
+ }
+ }
+ }) {
+ @Override public long getId() {
+ // Override getId to use all connections.
+ return tId.getAndIncrement();
+ }
+ };
+
+ threadsList.add(t0);
+
+ t0.start();
+ }
+
+ for (Thread t0 : threadsList)
+ t0.join();
assertTrue(latch.await(10, TimeUnit.SECONDS));
@@ -280,13 +328,13 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
@Override public boolean apply() {
Collection sessions = U.field(srv, "sessions");
- return sessions.size() == 2;
+ return sessions.size() == 2 * connectionsPerNode;
}
}, 5000);
Collection sessions = U.field(srv, "sessions");
- assertEquals(2, sessions.size());
+ assertEquals(2 * connectionsPerNode, sessions.size());
}
assertEquals(expMsgs, lsnr.cntr.get());
@@ -315,6 +363,7 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
spi.setIdleConnectionTimeout(60_000);
spi.setConnectTimeout(10_000);
spi.setSharedMemoryPort(-1);
+ spi.setConnectionsPerNode(connectionsPerNode);
return spi;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java
index b0353a6..5345a9b 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java
@@ -18,8 +18,6 @@
package org.apache.ignite.spi.communication.tcp;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.junits.spi.GridSpiAbstractConfigTest;
import org.apache.ignite.testframework.junits.spi.GridSpiTest;
@@ -47,6 +45,9 @@ public class GridTcpCommunicationSpiConfigSelfTest extends GridSpiAbstractConfig
checkNegativeSpiProperty(new TcpCommunicationSpi(), "ackSendThreshold", 0);
checkNegativeSpiProperty(new TcpCommunicationSpi(), "ackSendThreshold", -1);
checkNegativeSpiProperty(new TcpCommunicationSpi(), "unacknowledgedMessagesBufferSize", -1);
+ checkNegativeSpiProperty(new TcpCommunicationSpi(), "connectionsPerNode", 0);
+ checkNegativeSpiProperty(new TcpCommunicationSpi(), "connectionsPerNode", -1);
+ checkNegativeSpiProperty(new TcpCommunicationSpi(), "connectionsPerNode", Integer.MAX_VALUE);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
index f210bec..74be68d 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
@@ -109,7 +109,7 @@ public class GridTcpCommunicationSpiMultithreadedSelfTest extends GridSpiAbstrac
/**
* @param useShmem Use shared mem.
*/
- protected GridTcpCommunicationSpiMultithreadedSelfTest(boolean useShmem) {
+ GridTcpCommunicationSpiMultithreadedSelfTest(boolean useShmem) {
super(false);
this.useShmem = useShmem;
@@ -549,11 +549,17 @@ public class GridTcpCommunicationSpiMultithreadedSelfTest extends GridSpiAbstrac
}
for (CommunicationSpi spi : spis.values()) {
- final ConcurrentMap<UUID, GridCommunicationClient> clients = U.field(spi, "clients");
+ final ConcurrentMap<UUID, GridCommunicationClient[]> clients = U.field(spi, "clients");
assert GridTestUtils.waitForCondition(new PA() {
@Override public boolean apply() {
- return clients.isEmpty();
+ for (GridCommunicationClient[] clients0 : clients.values()) {
+ for (GridCommunicationClient client : clients0) {
+ if (client != null)
+ return false;
+ }
+ }
+ return true;
}
}, getTestTimeout()) : "Clients: " + clients;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/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 fb2dfd7..e0478da 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
@@ -361,6 +361,7 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
spi.setAckSendThreshold(ackCnt);
spi.setMessageQueueLimit(queueLimit);
spi.setSharedMemoryPort(-1);
+ spi.setConnectionsPerNode(1);
return spi;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryFailureDetectionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryFailureDetectionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryFailureDetectionSelfTest.java
index 95c9e40..b1aa119 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryFailureDetectionSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryFailureDetectionSelfTest.java
@@ -33,6 +33,7 @@ public class GridTcpCommunicationSpiRecoveryFailureDetectionSelfTest extends Gri
spi.setAckSendThreshold(5);
spi.setSocketSendBuffer(512);
spi.setSocketReceiveBuffer(512);
+ spi.setConnectionsPerNode(1);
return spi;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/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 d2b0ecd..91014dc 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
@@ -54,6 +54,7 @@ import org.eclipse.jetty.util.ConcurrentHashSet;
/**
*
*/
+@SuppressWarnings("unchecked")
@GridSpiTest(spi = TcpCommunicationSpi.class, group = "Communication SPI")
public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi> extends GridSpiAbstractTest<T> {
/** */
@@ -326,7 +327,12 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
assertTrue("Failed to wait for session close", ses0.closeTime() != 0);
- ses1.resumeReads().get();
+ try {
+ ses1.resumeReads().get();
+ }
+ catch (IgniteCheckedException ignore) {
+ // Can fail is ses1 was closed.
+ }
for (int j = 0; j < 100; j++) {
spi0.sendMessage(node1, new GridTestMessage(node0.id(), msgId.incrementAndGet(), 0));
@@ -436,12 +442,16 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
assertTrue("Failed to wait for session close", ses0.closeTime() != 0);
- ses1.resumeReads().get();
+ try {
+ ses1.resumeReads().get();
+ }
+ catch (IgniteCheckedException ignore) {
+ // Can fail is ses1 was closed.
+ }
// Wait when session is closed, then try to open new connection from node1.
GridTestUtils.waitForCondition(new GridAbsPredicate() {
- @Override
- public boolean apply() {
+ @Override public boolean apply() {
return ses1.closeTime() != 0;
}
}, awaitForSocketWriteTimeout());
@@ -553,7 +563,12 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
assertTrue("Failed to wait for session close", ses0.closeTime() != 0);
- ses1.resumeReads().get();
+ try {
+ ses1.resumeReads().get();
+ }
+ catch (IgniteCheckedException ignore) {
+ // Can fail is ses1 was closed.
+ }
sndFut.get();
@@ -600,6 +615,7 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
/**
* @param spi SPI.
+ * @param in {@code True} if need find inbound session.
* @return Session.
* @throws Exception If failed.
*/
@@ -642,6 +658,7 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
spi.setSocketWriteTimeout(1000);
spi.setSocketSendBuffer(512);
spi.setSocketReceiveBuffer(512);
+ spi.setConnectionsPerNode(1);
return spi;
}
@@ -739,9 +756,8 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
spi.spiStop();
}
- for (IgniteTestResources rsrcs : spiRsrcs) {
+ for (IgniteTestResources rsrcs : spiRsrcs)
rsrcs.stopThreads();
- }
spis.clear();
nodes.clear();
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/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 e153fe2..9928d93 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
@@ -397,6 +397,7 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
spi.setAckSendThreshold(ackCnt);
spi.setMessageQueueLimit(queueLimit);
spi.setSharedMemoryPort(-1);
+ spi.setConnectionsPerNode(1);
return spi;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/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 1d59a95..aafc5f9 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
@@ -633,6 +633,7 @@ public class HadoopExternalCommunication {
.logger(log.getLogger(GridNioServer.class))
.selectorCount(selectorsCnt)
.gridName(gridName)
+ .serverName("hadoop")
.tcpNoDelay(tcpNoDelay)
.directBuffer(directBuf)
.byteOrder(ByteOrder.nativeOrder())
http://git-wip-us.apache.org/repos/asf/ignite/blob/6023539d/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestBenchmark.java
----------------------------------------------------------------------
diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestBenchmark.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestBenchmark.java
new file mode 100644
index 0000000..bee45e0
--- /dev/null
+++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestBenchmark.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.yardstick.cache;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.yardstick.IgniteAbstractBenchmark;
+import org.yardstickframework.BenchmarkConfiguration;
+import org.yardstickframework.BenchmarkUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ *
+ */
+public class IgniteIoTestBenchmark extends IgniteAbstractBenchmark {
+ /** */
+ private List<ClusterNode> targetNodes;
+
+ /** */
+ private IgniteKernal ignite;
+
+ /** {@inheritDoc} */
+ @Override public void setUp(BenchmarkConfiguration cfg) throws Exception {
+ super.setUp(cfg);
+
+ ignite = (IgniteKernal)ignite();
+
+ targetNodes = new ArrayList<>();
+
+ ClusterNode loc = ignite().cluster().localNode();
+
+ Collection<ClusterNode> nodes = ignite().cluster().forServers().nodes();
+
+ for (ClusterNode node : nodes) {
+ if (!loc.equals(node))
+ targetNodes.add(node);
+ }
+
+ if (targetNodes.isEmpty())
+ throw new IgniteException("Failed to find remote server nodes [nodes=" + nodes + ']');
+
+ BenchmarkUtils.println(cfg, "Initialized target nodes: " + targetNodes + ']');
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean test(Map<Object, Object> ctx) throws Exception {
+ ClusterNode node = targetNodes.get(nextRandom(targetNodes.size()));
+
+ ignite.sendIoTest(node, null, false).get();
+
+ return true;
+ }
+}