added turbo debug mode to communication
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/6f48d4a9 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/6f48d4a9 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/6f48d4a9 Branch: refs/heads/ignite-1189 Commit: 6f48d4a978036f571ca37a5388d9da6449f01114 Parents: 32f84c1 Author: Yakov Zhdanov <yzhda...@gridgain.com> Authored: Fri Jul 24 19:50:28 2015 +0300 Committer: Yakov Zhdanov <yzhda...@gridgain.com> Committed: Fri Jul 24 19:50:28 2015 +0300 ---------------------------------------------------------------------- .../org/apache/ignite/internal/IgnitionEx.java | 26 ++++++ .../managers/communication/GridIoManager.java | 84 +++++++++++++++++++- .../src/test/config/io-manager-benchmark.xml | 3 +- 3 files changed, 110 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6f48d4a9/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java index 5cbe377..73de99a 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java @@ -1087,6 +1087,32 @@ public class IgnitionEx { } /** + * Gets grid instance without waiting its initialization and not throwing any exception. + * + * @param locNodeId ID of local node the requested grid instance is managing. + * @return Grid instance or {@code null}. + */ + public static IgniteKernal gridxx(UUID locNodeId) { + IgniteNamedInstance dfltGrid0 = dfltGrid; + + if (dfltGrid0 != null) { + IgniteKernal g = dfltGrid0.grid(); + + if (g != null && g.getLocalNodeId().equals(locNodeId)) + return g; + } + + for (IgniteNamedInstance grid : grids.values()) { + IgniteKernal g = grid.grid(); + + if (g != null && g.getLocalNodeId().equals(locNodeId)) + return g; + } + + return null; + } + + /** * Gets an named grid instance. If grid name is {@code null} or empty string, * then default no-name grid will be returned. Note that caller of this method * should not assume that it will return the same instance every time. http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6f48d4a9/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 c1fb79a..65b6fad 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 @@ -57,6 +57,9 @@ import static org.jsr166.ConcurrentLinkedHashMap.QueuePolicy.*; * Grid communication manager. */ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializable>> { + /** */ + public static volatile boolean TURBO_DEBUG_MODE; + /** Empty array of message factories. */ public static final MessageFactory[] EMPTY = {}; @@ -894,7 +897,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa if (msgC == null) { // Message from local node can be processed in sync manner. - assert locNodeId.equals(nodeId); + assert locNodeId.equals(nodeId) || TURBO_DEBUG_MODE; unwindMessageSet(set, lsnr); @@ -1019,6 +1022,85 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa } /** + * This method can be used for debugging tricky concurrency issues + * with multi-nodes in single JVM. + * <p> + * This method eliminates network between nodes started in single JVM + * when {@link #TURBO_DEBUG_MODE} is set to {@code true}. + * <p> + * How to use it: + * <ol> + * <li>Replace {@link #send(ClusterNode, Object, int, Message, byte, boolean, long, boolean)} + * with this method.</li> + * <li>Start all grids for your test, then set {@link #TURBO_DEBUG_MODE} to {@code true}.</li> + * <li>Perform test operations on the topology. No network will be there.</li> + * <li>DO NOT turn on turbo debug before all grids started. This will cause deadlocks.</li> + * </ol> + * + * @param node Destination node. + * @param topic Topic to send the message to. + * @param topicOrd GridTopic enumeration ordinal. + * @param msg Message to send. + * @param plc Type of processing. + * @param ordered Ordered flag. + * @param timeout Timeout. + * @param skipOnTimeout Whether message can be skipped on timeout. + * @throws IgniteCheckedException Thrown in case of any errors. + */ + private void sendTurboDebug( + ClusterNode node, + Object topic, + int topicOrd, + Message msg, + byte plc, + boolean ordered, + long timeout, + boolean skipOnTimeout + ) throws IgniteCheckedException { + assert node != null; + assert topic != null; + assert msg != null; + + GridIoMessage ioMsg = new GridIoMessage(plc, topic, topicOrd, msg, ordered, timeout, skipOnTimeout); + + IgniteKernal rmt; + + if (locNodeId.equals(node.id())) { + assert plc != P2P_POOL; + + CommunicationListener commLsnr = this.commLsnr; + + if (commLsnr == null) + throw new IgniteCheckedException("Trying to send message when grid is not fully started."); + + if (ordered) + processOrderedMessage(locNodeId, ioMsg, plc, null); + else + processRegularMessage0(ioMsg, locNodeId); + } + else if (TURBO_DEBUG_MODE && (rmt = IgnitionEx.gridxx(locNodeId)) != null) { + if (ioMsg.isOrdered()) + rmt.context().io().processOrderedMessage(locNodeId, ioMsg, ioMsg.policy(), null); + else + rmt.context().io().processRegularMessage0(ioMsg, locNodeId); + } + else { + if (topicOrd < 0) + ioMsg.topicBytes(marsh.marshal(topic)); + + try { + getSpi().sendMessage(node, ioMsg); + } + catch (IgniteSpiException e) { + throw new IgniteCheckedException("Failed to send message (node may have left the grid or " + + "TCP connection cannot be established due to firewall issues) " + + "[node=" + node + ", topic=" + topic + + ", msg=" + msg + ", policy=" + plc + ']', e); + } + } + } + + /** * @param nodeId Id of destination node. * @param topic Topic to send the message to. * @param msg Message to send. http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6f48d4a9/modules/core/src/test/config/io-manager-benchmark.xml ---------------------------------------------------------------------- diff --git a/modules/core/src/test/config/io-manager-benchmark.xml b/modules/core/src/test/config/io-manager-benchmark.xml index 9b97407..4f481ec 100644 --- a/modules/core/src/test/config/io-manager-benchmark.xml +++ b/modules/core/src/test/config/io-manager-benchmark.xml @@ -27,8 +27,7 @@ <property name="communicationSpi"> <bean class="org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi"> - <property name="asyncSend" value="true"/> - <property name="tcpNoDelay" value="false"/> + <property name="sharedMemoryPort" value="-1"/> </bean> </property>
