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 2017/06/10 06:20:33 UTC
[02/19] ignite git commit: IGNITE-5155
IGNITE-5155
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/e3dd573d
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/e3dd573d
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/e3dd573d
Branch: refs/heads/ignite-5155
Commit: e3dd573df4d7fd1d95035ef81cce475bab3cc1dc
Parents: 3ab4b9a
Author: Igor Seliverstov <gv...@gmail.com>
Authored: Wed Jun 7 18:56:04 2017 +0300
Committer: Igor Seliverstov <gv...@gmail.com>
Committed: Wed Jun 7 18:56:04 2017 +0300
----------------------------------------------------------------------
.../apache/ignite/IgniteSystemProperties.java | 6 +
.../org/apache/ignite/internal/GridTopic.java | 5 +-
.../ignite/internal/IgniteDiagnosticAware.java | 25 +
.../internal/IgniteDiagnosticMessage.java | 482 +++++++++++++++++++
.../apache/ignite/internal/IgniteKernal.java | 7 +-
.../communication/GridIoMessageFactory.java | 8 +-
.../processors/cache/GridCacheIoManager.java | 39 ++
.../GridCachePartitionExchangeManager.java | 9 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 30 +-
.../dht/GridPartitionedSingleGetFuture.java | 24 +-
.../colocated/GridDhtColocatedLockFuture.java | 44 +-
.../GridDhtPartitionsExchangeFuture.java | 39 +-
.../processors/cluster/ClusterProcessor.java | 354 +++++++++++++-
.../ignite/internal/util/nio/GridNioServer.java | 217 ++++++---
.../communication/tcp/TcpCommunicationSpi.java | 167 ++++---
...agnosticMessagesMultipleConnectionsTest.java | 35 ++
.../managers/IgniteDiagnosticMessagesTest.java | 152 ++++++
17 files changed, 1517 insertions(+), 126 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index 84f3732..539f288 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -448,6 +448,12 @@ public final class IgniteSystemProperties {
/** If this property is set to {@code true} then Ignite will log thread dump in case of partition exchange timeout. */
public static final String IGNITE_THREAD_DUMP_ON_EXCHANGE_TIMEOUT = "IGNITE_THREAD_DUMP_ON_EXCHANGE_TIMEOUT";
+ /** */
+ public static final String IGNITE_IO_DUMP_ON_TIMEOUT = "IGNITE_IO_DUMP_ON_TIMEOUT";
+
+ /** */
+ public static final String IGNITE_DIAGNOSTIC_ENABLED = "IGNITE_DIAGNOSTIC_ENABLED";
+
/** Cache operations that take more time than value of this property will be output to log. Set to {@code 0} to disable. */
public static final String IGNITE_LONG_OPERATIONS_DUMP_TIMEOUT = "IGNITE_LONG_OPERATIONS_DUMP_TIMEOUT";
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/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 c382999..ab2d572 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
@@ -111,7 +111,10 @@ public enum GridTopic {
TOPIC_METADATA_REQ,
/** */
- TOPIC_SCHEMA;
+ TOPIC_SCHEMA,
+
+ /** */
+ TOPIC_INTERNAL_DIAGNOSTIC;
/** Enum values. */
private static final GridTopic[] VALS = values();
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticAware.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticAware.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticAware.java
new file mode 100644
index 0000000..44f531e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticAware.java
@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+/**
+ *
+ */
+public interface IgniteDiagnosticAware {
+ public void dumpDiagnosticInfo();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticMessage.java
new file mode 100644
index 0000000..75c5efc
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticMessage.java
@@ -0,0 +1,482 @@
+/*
+ * 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 java.nio.ByteBuffer;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
+import org.apache.ignite.internal.processors.cache.GridCachePartitionExchangeManager;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.future.GridFinishedFuture;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+
+/**
+ *
+ */
+public class IgniteDiagnosticMessage implements Message {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private static final ThreadLocal<DateFormat> dateFormat = new ThreadLocal<DateFormat>() {
+ @Override protected DateFormat initialValue() {
+ return new SimpleDateFormat("HH:mm:ss.SSS");
+ }
+ };
+
+ /** */
+ private long futId;
+
+ /** */
+ private String msg;
+
+ /** */
+ private byte[] cBytes;
+
+ /**
+ * Required by {@link GridIoMessageFactory}.
+ */
+ @SuppressWarnings("WeakerAccess")
+ public IgniteDiagnosticMessage() {
+ // No-op.
+ }
+
+ /**
+ * @param ctx Context.
+ * @param c Closure to run.
+ * @param futId Future ID.
+ * @return Request message.
+ * @throws IgniteCheckedException If failed.
+ */
+ public static IgniteDiagnosticMessage createRequest(GridKernalContext ctx,
+ IgniteClosure<GridKernalContext, String> c,
+ long futId)
+ throws IgniteCheckedException
+ {
+ byte[] cBytes = U.marshal(ctx.config().getMarshaller(), c);
+
+ IgniteDiagnosticMessage msg = new IgniteDiagnosticMessage();
+
+ msg.futId = futId;
+ msg.cBytes = cBytes;
+
+ return msg;
+ }
+
+ /**
+ * @param msg0 Message.
+ * @param futId Future ID.
+ * @return Response message.
+ */
+ public static IgniteDiagnosticMessage createResponse(String msg0, long futId) {
+ IgniteDiagnosticMessage msg = new IgniteDiagnosticMessage();
+
+ msg.futId = futId;
+ msg.msg = msg0;
+
+ return msg;
+ }
+
+ /**
+ * @param ctx Context.
+ * @return Unmarshalled closure.
+ * @throws IgniteCheckedException If failed.
+ */
+ public IgniteClosure<GridKernalContext, String> unmarshalClosure(GridKernalContext ctx)
+ throws IgniteCheckedException {
+ assert cBytes != null;
+
+ return U.unmarshal(ctx, cBytes, null);
+ }
+
+ /**
+ * @return Future ID.
+ */
+ public long futureId() {
+ return futId;
+ }
+
+ /**
+ * @return {@code True} if this is request message.
+ */
+ public boolean request() {
+ return cBytes != null;
+ }
+
+ /**
+ * @return Message string.
+ */
+ public String message() {
+ return msg;
+ }
+
+ /** {@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.writeByteArray("cBytes", cBytes))
+ return false;
+
+ writer.incrementState();
+
+ case 1:
+ if (!writer.writeLong("futId", futId))
+ return false;
+
+ writer.incrementState();
+
+ case 2:
+ if (!writer.writeString("msg", msg))
+ 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:
+ cBytes = reader.readByteArray("cBytes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 1:
+ futId = reader.readLong("futId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 2:
+ msg = reader.readString("msg");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ }
+
+ return reader.afterMessageRead(IgniteDiagnosticMessage.class);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return -55;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 3;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAckReceived() {
+ // No-op.
+ }
+
+ /**
+ *
+ */
+ public static class BaseClosure implements IgniteClosure<GridKernalContext, String> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ protected final UUID nodeId;
+
+ /**
+ * @param ctx Local node context.
+ */
+ public BaseClosure(GridKernalContext ctx) {
+ this.nodeId = ctx.localNodeId();
+ }
+
+ /** {@inheritDoc} */
+ @Override public final String apply(GridKernalContext ctx) {
+ try {
+ IgniteInternalFuture<String> commInfo = dumpCommunicationInfo(ctx, nodeId);
+
+ StringBuilder sb = new StringBuilder();
+
+ dumpNodeBasicInfo(sb, ctx);
+
+ sb.append(U.nl());
+
+ dumpExchangeInfo(sb, ctx);
+
+ sb.append(U.nl()).append(commInfo.get());
+
+ moreInfo(sb, ctx);
+
+ return sb.toString();
+ }
+ catch (Exception e) {
+ ctx.cluster().diagnosticLog().error("Failed to execute diagnostic message closure: " + e, e);
+
+ return "Failed to execute diagnostic message closure: " + e;
+ }
+ }
+
+ /**
+ * @param sb String builder.
+ * @param ctx Context.
+ */
+ protected void moreInfo(StringBuilder sb, GridKernalContext ctx) {
+ // No-op.
+ }
+ }
+
+ /**
+ *
+ */
+ public static class TxEntriesInfoClosure extends BaseClosure {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final int cacheId;
+
+ /** */
+ private final Collection<KeyCacheObject> keys;
+
+ /**
+ * @param ctx Context.
+ * @param cacheId Cache ID.
+ * @param keys Keys.
+ */
+ public TxEntriesInfoClosure(GridKernalContext ctx, int cacheId, Collection<KeyCacheObject> keys) {
+ super(ctx);
+
+ this.cacheId = cacheId;
+ this.keys = keys;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void moreInfo(StringBuilder sb, GridKernalContext ctx) {
+ sb.append(U.nl());
+
+ GridCacheContext cctx = ctx.cache().context().cacheContext(cacheId);
+
+ if (cctx == null) {
+ sb.append("Failed to find cache with id: ").append(cacheId);
+
+ return;
+ }
+
+ try {
+ for (KeyCacheObject key : keys)
+ key.finishUnmarshal(cctx.cacheObjectContext(), null);
+ }
+ catch (IgniteCheckedException e) {
+ ctx.cluster().diagnosticLog().error("Failed to unmarshal key: " + e, e);
+
+ sb.append("Failed to unmarshal key: ").append(e).append(U.nl());
+ }
+
+ sb.append("Cache entries [cacheId=").append(cacheId)
+ .append(", cacheName=").append(cctx.name()).append("]: ");
+
+ for (KeyCacheObject key : keys) {
+ GridCacheMapEntry e = (GridCacheMapEntry)cctx.cache().peekEx(key);
+
+ sb.append(U.nl()).append(" Key [key=").append(key).append(", entry=").append(e).append("]");
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ public static class ExchangeInfoClosure extends BaseClosure {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final AffinityTopologyVersion topVer;
+
+ /**
+ * @param ctx Context.
+ * @param topVer Exchange version.
+ */
+ public ExchangeInfoClosure(GridKernalContext ctx, AffinityTopologyVersion topVer) {
+ super(ctx);
+
+ this.topVer = topVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void moreInfo(StringBuilder sb, GridKernalContext ctx) {
+ sb.append(U.nl());
+
+ List<GridDhtPartitionsExchangeFuture> futs = ctx.cache().context().exchange().exchangeFutures();
+
+ for (GridDhtPartitionsExchangeFuture fut : futs) {
+ if (topVer.equals(fut.topologyVersion())) {
+ sb.append("Exchange future: ").append(fut);
+
+ return;
+ }
+ }
+
+ sb.append("Failed to find exchange future: ").append(topVer);
+ }
+ }
+
+ /**
+ *
+ */
+ public static class TxInfoClosure extends BaseClosure {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final GridCacheVersion dhtVer;
+
+ /** */
+ private final GridCacheVersion nearVer;
+
+ /**
+ * @param ctx Context.
+ * @param dhtVer Tx dht version.
+ * @param nearVer Tx near version.
+ */
+ public TxInfoClosure(GridKernalContext ctx,
+ GridCacheVersion dhtVer,
+ GridCacheVersion nearVer) {
+ super(ctx);
+
+ this.dhtVer = dhtVer;
+ this.nearVer = nearVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void moreInfo(StringBuilder sb, GridKernalContext ctx) {
+ sb.append(U.nl())
+ .append("Related transactions [dhtVer=").append(dhtVer)
+ .append(", nearVer=").append(nearVer).append("]: ");
+
+ boolean found = false;
+
+ for (IgniteInternalTx tx : ctx.cache().context().tm().activeTransactions()) {
+ if (dhtVer.equals(tx.xidVersion()) || nearVer.equals(tx.nearXidVersion())) {
+ sb.append(U.nl())
+ .append(" [ver=").append(tx.xidVersion())
+ .append(", nearVer=").append(tx.nearXidVersion())
+ .append(", topVer=").append(tx.topologyVersion())
+ .append(", state=").append(tx.state())
+ .append(", fullTx=").append(tx).append(']');
+
+ found = true;
+ }
+ }
+
+ if (!found)
+ sb.append(U.nl()).append("Failed to find related transactions.");
+ }
+ }
+
+ /**
+ *
+ * @param sb String builder.
+ * @param ctx Context.
+ */
+ static void dumpNodeBasicInfo(StringBuilder sb, GridKernalContext ctx) {
+ sb.append("General node info [id=").append(ctx.localNodeId())
+ .append(", client=").append(ctx.clientNode())
+ .append(", discoTopVer=").append(ctx.discovery().topologyVersionEx())
+ .append(", time=").append(formatTime(U.currentTimeMillis())).append(']');
+ }
+
+ /**
+ * @param sb String builder.
+ * @param ctx Context.
+ */
+ static void dumpExchangeInfo(StringBuilder sb, GridKernalContext ctx) {
+ GridCachePartitionExchangeManager exchMgr = ctx.cache().context().exchange();
+ GridDhtTopologyFuture fut = exchMgr.lastTopologyFuture();
+
+ sb.append("Partitions exchange info [readyVer=").append(exchMgr.readyAffinityVersion()).append(']').append(U.nl())
+ .append("Last initialized exchange future: ").append(fut);
+ }
+
+ /**
+ * @param ctx Context.
+ * @param nodeId Target node ID.
+ * @return Communication information future.
+ */
+ public static IgniteInternalFuture<String> dumpCommunicationInfo(GridKernalContext ctx, UUID nodeId) {
+ if (ctx.config().getCommunicationSpi() instanceof TcpCommunicationSpi)
+ return ((TcpCommunicationSpi) ctx.config().getCommunicationSpi()).dumpNodeStatistics(nodeId);
+ else
+ return new GridFinishedFuture<>("Unexpected communication SPI: " + ctx.config().getCommunicationSpi());
+ }
+ /**
+ * @param time Time.
+ * @return Time string.
+ */
+ private static String formatTime(long time) {
+ return dateFormat.get().format(new Date(time));
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(IgniteDiagnosticMessage.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 6d05147..7757644 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -950,7 +950,12 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
// Start platform plugins.
if (ctx.config().getPlatformConfiguration() != null)
- startProcessor(new PlatformPluginProcessor(ctx));fillNodeAttributes(clusterProc.updateNotifierEnabled());}
+ startProcessor(new PlatformPluginProcessor(ctx));
+
+ ctx.cluster().initListeners();
+
+ fillNodeAttributes(clusterProc.updateNotifierEnabled());
+ }
catch (Throwable e) {
U.error(
log, "Exception during start processors, node will be stopped and close connections", e);
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/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 753d8af..12f160b 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
@@ -26,6 +26,7 @@ import org.apache.ignite.internal.GridJobSiblingsRequest;
import org.apache.ignite.internal.GridJobSiblingsResponse;
import org.apache.ignite.internal.GridTaskCancelRequest;
import org.apache.ignite.internal.GridTaskSessionRequest;
+import org.apache.ignite.internal.IgniteDiagnosticMessage;
import org.apache.ignite.internal.binary.BinaryEnumObjectImpl;
import org.apache.ignite.internal.binary.BinaryObjectImpl;
import org.apache.ignite.internal.managers.checkpoint.GridCheckpointRequest;
@@ -178,6 +179,11 @@ public class GridIoMessageFactory implements MessageFactory {
Message msg = null;
switch (type) {
+ case -55:
+ msg = new IgniteDiagnosticMessage();
+
+ break;
+
// -54 is reserved for SQL.
case -53:
@@ -875,7 +881,7 @@ public class GridIoMessageFactory implements MessageFactory {
break;
- // [-3..119] [124..127] [-23..-27] [-36..-47]- this
+ // [-3..119] [124..127] [-23..-27] [-36..-55]- this
// [120..123] - DR
// [-4..-22, -30..-35] - SQL
default:
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index a251047..e66b09e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -17,11 +17,15 @@
package org.apache.ignite.internal.processors.cache;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
@@ -126,6 +130,29 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
/** Deployment enabled. */
private boolean depEnabled;
+ /** */
+ private static final int MAX_STORED_PENDING_MESSAGES = 100;
+
+ /** */
+ private final List<GridCacheMessage> pendingMsgs = new ArrayList<>(MAX_STORED_PENDING_MESSAGES);
+
+ /**
+ *
+ */
+ public void dumpPendingMessages() {
+ synchronized (pendingMsgs) {
+ if (pendingMsgs.isEmpty())
+ return;
+
+ log.info("Pending cache messages waiting for exchange [" +
+ "readyVer=" + cctx.exchange().readyAffinityVersion() +
+ ", discoVer=" + cctx.discovery().topologyVersion() + ']');
+
+ for (GridCacheMessage msg : pendingMsgs)
+ log.info("Message [waitVer=" + msg.topologyVersion() + ", msg=" + msg + ']');
+ }
+ }
+
/** Message listener. */
private GridMessageListener lsnr = new GridMessageListener() {
@Override public void onMessage(final UUID nodeId, final Object msg) {
@@ -214,6 +241,14 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
}
if (fut != null && !fut.isDone()) {
+ synchronized (pendingMsgs) {
+ if (pendingMsgs.size() == MAX_STORED_PENDING_MESSAGES)
+ // remove oldest
+ pendingMsgs.remove(0);
+
+ pendingMsgs.add(cacheMsg);
+ }
+
Thread curThread = Thread.currentThread();
final int stripe = curThread instanceof IgniteThread ? ((IgniteThread)curThread).stripe() : -1;
@@ -222,6 +257,10 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
@Override public void apply(IgniteInternalFuture<?> t) {
Runnable c = new Runnable() {
@Override public void run() {
+ synchronized (pendingMsgs) {
+ pendingMsgs.remove(cacheMsg);
+ }
+
IgniteLogger log = cacheMsg.messageLogger(cctx);
if (log.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 2be4d0f..13b5644 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -1389,7 +1389,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
int cnt = 0;
for (GridDhtPartitionsExchangeFuture fut : exchFuts.values()) {
- U.warn(log, ">>> " + fut);
+ U.warn(log, ">>> " + fut.shortInfo());
if (++cnt == 10)
break;
@@ -1403,8 +1403,11 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
cctx.affinity().dumpDebugInfo();
+ cctx.io().dumpPendingMessages();
+
// Dump IO manager statistics.
- cctx.gridIO().dumpStats();
+ if (IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_IO_DUMP_ON_TIMEOUT, false))
+ cctx.gridIO().dumpStats();
}
/**
@@ -1698,7 +1701,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
for (CachePartitionExchangeWorkerTask task: futQ) {
if (isExchangeTask(task))
- U.warn(log, ">>> " + task);
+ U.warn(log, ">>> " + ((GridDhtPartitionsExchangeFuture)task).shortInfo());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index 75f8366..8bb6111 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -36,6 +36,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteDiagnosticAware;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -100,7 +101,7 @@ import static org.apache.ignite.transactions.TransactionState.PREPARED;
*/
@SuppressWarnings("unchecked")
public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<IgniteInternalTx, GridNearTxPrepareResponse>
- implements GridCacheMvccFuture<GridNearTxPrepareResponse> {
+ implements GridCacheMvccFuture<GridNearTxPrepareResponse>, IgniteDiagnosticAware {
/** */
private static final long serialVersionUID = 0L;
@@ -1567,6 +1568,33 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
}
/** {@inheritDoc} */
+ @Override public void dumpDiagnosticInfo() {
+ if (!isDone()) {
+ for (IgniteInternalFuture fut : futures()) {
+ if (!fut.isDone() && fut instanceof MiniFuture) {
+ MiniFuture f = (MiniFuture)fut;
+
+ if (!f.node().isLocal()) {
+ GridCacheVersion dhtVer = tx.xidVersion();
+ GridCacheVersion nearVer = tx.nearXidVersion();
+
+ cctx.kernalContext().cluster().dumpRemoteTxInfo(f.nodeId, dhtVer, nearVer, "GridDhtTxPrepareFuture " +
+ "waiting for response [node=" + f.nodeId +
+ ", topVer=" + tx.topologyVersion() +
+ ", dhtVer=" + dhtVer +
+ ", nearVer=" + nearVer +
+ ", futId=" + futId +
+ ", miniId=" + f.futId +
+ ", tx=" + tx + ']');
+
+ return;
+ }
+ }
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
@Override public String apply(IgniteInternalFuture<?> f) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 3f612f7..9ce3842 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteDiagnosticAware;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
@@ -63,7 +64,7 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDh
*
*/
public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Object> implements GridCacheFuture<Object>,
- CacheGetFuture {
+ CacheGetFuture, IgniteDiagnosticAware {
/** Logger reference. */
private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
@@ -767,6 +768,27 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
}
/** {@inheritDoc} */
+ @Override public void dumpDiagnosticInfo() {
+ if (!isDone()) {
+ UUID nodeId;
+ AffinityTopologyVersion topVer;
+
+ synchronized (this) {
+ nodeId = node != null ? node.id() : null;
+ topVer = this.topVer;
+ }
+
+ if (nodeId != null)
+ cctx.kernalContext().cluster().dumpBasicInfo(nodeId, "GridPartitionedSingleGetFuture waiting for " +
+ "response [node=" + nodeId +
+ ", key=" + key +
+ ", futId=" + futId +
+ ", topVer=" + topVer + ']',
+ null);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridPartitionedSingleGetFuture.class, this);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
index 4442b3a..69bd57f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
@@ -30,6 +30,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteDiagnosticAware;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
@@ -83,7 +84,7 @@ import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ;
* Colocated cache lock future.
*/
public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityFuture<Boolean>
- implements GridCacheMvccFuture<Boolean> {
+ implements GridCacheMvccFuture<Boolean>, IgniteDiagnosticAware {
/** */
private static final long serialVersionUID = 0L;
@@ -596,13 +597,51 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
}
/** {@inheritDoc} */
+ @Override public void dumpDiagnosticInfo() {
+ if (!isDone()) {
+ for (IgniteInternalFuture fut : futures()) {
+ if (!fut.isDone() && isMini(fut)) {
+ MiniFuture m = (MiniFuture)fut;
+
+ AffinityTopologyVersion topVer = null;
+ UUID rmtNodeId = null;
+
+ synchronized (m) {
+ if (!m.rcvRes && !m.node.isLocal()) {
+ rmtNodeId = m.node.id();
+
+ topVer = this.topVer;
+ }
+ }
+
+ if (rmtNodeId != null) {
+ cctx.kernalContext().cluster().dumpTxKeyInfo(rmtNodeId, cctx.cacheId(), m.keys,
+ "GridDhtColocatedLockFuture waiting for response [node=" + rmtNodeId +
+ ", cache=" + cctx.name() +
+ ", miniId=" + m.futId +
+ ", topVer=" + topVer +
+ ", keys=" + m.keys + ']');
+
+ return;
+ }
+ }
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
@Override public String apply(IgniteInternalFuture<?> f) {
if (isMini(f)) {
MiniFuture m = (MiniFuture)f;
- return "[node=" + m.node().id() + ", loc=" + m.node().isLocal() + ", done=" + f.isDone() + "]";
+ synchronized (m) {
+ return "[node=" + m.node().id() +
+ ", rcvRes=" + m.rcvRes +
+ ", loc=" + m.node().isLocal() +
+ ", done=" + f.isDone() + "]";
+ }
}
else
return "[loc=true, done=" + f.isDone() + "]";
@@ -610,6 +649,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
});
return S.toString(GridDhtColocatedLockFuture.class, this,
+ "topVer", topVer,
"innerFuts", futs,
"inTx", inTx(),
"super", super.toString());
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 26406c6..88d31e2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -42,6 +42,7 @@ import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
+import org.apache.ignite.internal.IgniteDiagnosticAware;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
@@ -108,7 +109,7 @@ import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYS
*/
@SuppressWarnings({"TypeMayBeWeakened", "unchecked"})
public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityTopologyVersion>
- implements Comparable<GridDhtPartitionsExchangeFuture>, GridDhtTopologyFuture, CachePartitionExchangeWorkerTask {
+ implements Comparable<GridDhtPartitionsExchangeFuture>, GridDhtTopologyFuture, CachePartitionExchangeWorkerTask, IgniteDiagnosticAware {
/** Dummy flag. */
private final boolean dummy;
@@ -2157,6 +2158,42 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
}
/** {@inheritDoc} */
+ @Override public void dumpDiagnosticInfo() {
+ if (!isDone()) {
+ ClusterNode crd;
+ Set<UUID> remaining;
+
+ synchronized (this) {
+ crd = this.crd;
+ remaining = new HashSet<>(this.remaining);
+ }
+
+ if (crd != null) {
+ if (!crd.isLocal()) {
+ cctx.kernalContext().cluster().dumpExchangeInfo(crd.id(), topologyVersion(), "Exchange future waiting for coordinator " +
+ "response [crd=" + crd.id() + ", topVer=" + topologyVersion() + ']');
+ }
+ else if (!remaining.isEmpty()){
+ UUID nodeId = remaining.iterator().next();
+
+ cctx.kernalContext().cluster().dumpExchangeInfo(crd.id(), topologyVersion(), "Exchange future waiting for server " +
+ "response [node=" + nodeId + ", topVer=" + topologyVersion() + ']');
+ }
+ }
+ }
+ }
+
+ /**
+ * @return Short information string.
+ */
+ public String shortInfo() {
+ return "GridDhtPartitionsExchangeFuture [topVer=" + topologyVersion() +
+ ", evt=" + (discoEvt != null ? discoEvt.type() : -1) +
+ ", evtNode=" + (discoEvt != null ? discoEvt.eventNode() : null) +
+ ", done=" + isDone() + ']';
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
Set<UUID> remaining;
List<ClusterNode> srvNodes;
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java
index 542ee43..dcde202 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java
@@ -24,24 +24,47 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
+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.IgniteDiagnosticMessage;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteKernal;
-import org.apache.ignite.internal.IgniteProperties;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.IgniteClusterImpl;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.GridTimerTask;
+import org.apache.ignite.internal.util.future.GridFinishedFuture;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.future.IgniteFinishedFutureImpl;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.CI1;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.spi.discovery.DiscoveryDataBag;
import org.apache.ignite.spi.discovery.DiscoveryDataBag.GridDiscoveryData;
import org.jetbrains.annotations.Nullable;
-import static org.apache.ignite.IgniteSystemProperties.IGNITE_UPDATE_NOTIFIER;
+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.GridComponent.DiscoveryDataExchangeType.CLUSTER_PROC;
import static org.apache.ignite.internal.IgniteVersionUtils.VER_STR;
@@ -50,6 +73,13 @@ import static org.apache.ignite.internal.IgniteVersionUtils.VER_STR;
*/
public class ClusterProcessor extends GridProcessorAdapter {
/** */
+ private final boolean DIAGNOSTIC_ENABLED =
+ IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_DIAGNOSTIC_ENABLED, false);
+
+ /** */
+ private static final String DIAGNOSTIC_LOG_CATEGORY = "org.apache.ignite.internal.diagnostic";
+
+ /** */
private static final String ATTR_UPDATE_NOTIFIER_STATUS = "UPDATE_NOTIFIER_STATUS";
/** Periodic version check delay. */
@@ -72,6 +102,16 @@ public class ClusterProcessor extends GridProcessorAdapter {
@GridToStringExclude
private GridUpdateNotifier verChecker;
+ /** */
+ private final IgniteLogger diagnosticLog;
+
+ /** */
+ private final AtomicReference<ConcurrentHashMap<Long, InternalDiagnosticFuture>> diagnosticFutMap =
+ new AtomicReference<>();
+
+ /** */
+ private final AtomicLong diagFutId = new AtomicLong();
+
/**
* @param ctx Kernal context.
*/
@@ -79,6 +119,102 @@ public class ClusterProcessor extends GridProcessorAdapter {
super(ctx);
cluster = new IgniteClusterImpl(ctx);
+
+ diagnosticLog = ctx.log(DIAGNOSTIC_LOG_CATEGORY);
+ }
+
+ /**
+ * @throws IgniteCheckedException If failed.
+ */
+ public void initListeners() throws IgniteCheckedException {
+ ctx.event().addLocalEventListener(new GridLocalEventListener() {
+ @Override public void onEvent(Event evt) {
+ assert evt instanceof DiscoveryEvent;
+ assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT;
+
+ DiscoveryEvent discoEvt = (DiscoveryEvent)evt;
+
+ UUID nodeId = discoEvt.eventNode().id();
+
+ ConcurrentHashMap<Long, InternalDiagnosticFuture> futs = diagnosticFutMap.get();
+
+ if (futs != null) {
+ for (InternalDiagnosticFuture fut : futs.values()) {
+ if (fut.nodeId.equals(nodeId))
+ fut.onDone("Target node failed: " + nodeId);
+ }
+ }
+ }
+ },
+ EVT_NODE_FAILED, EVT_NODE_LEFT);
+
+ ctx.io().addMessageListener(GridTopic.TOPIC_INTERNAL_DIAGNOSTIC, new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ if (msg instanceof IgniteDiagnosticMessage) {
+ IgniteDiagnosticMessage msg0 = (IgniteDiagnosticMessage)msg;
+
+ if (msg0.request()) {
+ ClusterNode node = ctx.discovery().node(nodeId);
+
+ if (node == null) {
+ if (diagnosticLog.isDebugEnabled()) {
+ diagnosticLog.debug("Skip diagnostic request, sender node left " +
+ "[node=" + nodeId + ", msg=" + msg + ']');
+ }
+
+ return;
+ }
+
+ String resMsg;
+
+ IgniteClosure<GridKernalContext, String> c;
+
+ try {
+ c = msg0.unmarshalClosure(ctx);
+
+ resMsg = c.apply(ctx);
+ }
+ catch (Exception e) {
+ U.error(diagnosticLog, "Failed to run diagnostic closure: " + e, e);
+
+ resMsg = "Failed to run diagnostic closure: " + e;
+ }
+
+ IgniteDiagnosticMessage res = IgniteDiagnosticMessage.createResponse(resMsg, msg0.futureId());
+
+ try {
+ ctx.io().sendToGridTopic(node, GridTopic.TOPIC_INTERNAL_DIAGNOSTIC, res, GridIoPolicy.SYSTEM_POOL);
+ }
+ catch (ClusterTopologyCheckedException e) {
+ if (diagnosticLog.isDebugEnabled()) {
+ diagnosticLog.debug("Failed to send diagnostic response, node left " +
+ "[node=" + nodeId + ", msg=" + msg + ']');
+ }
+ }
+ catch (IgniteCheckedException e) {
+ U.error(diagnosticLog, "Failed to send diagnostic response [msg=" + msg0 + "]", e);
+ }
+ }
+ else {
+ InternalDiagnosticFuture fut = diagnosticFuturesMap().get(msg0.futureId());
+
+ if (fut != null)
+ fut.onResponse(msg0);
+ else
+ U.warn(diagnosticLog, "Failed to find diagnostic message future [msg=" + msg0 + ']');
+ }
+ }
+ else
+ U.warn(diagnosticLog, "Received unexpected message: " + msg);
+ }
+ });
+ }
+
+ /**
+ * @return Logger for diagnostic category.
+ */
+ public IgniteLogger diagnosticLog() {
+ return diagnosticLog;
}
/**
@@ -212,6 +348,178 @@ public class ClusterProcessor extends GridProcessorAdapter {
}
/**
+ * @param nodeId Target node ID.
+ * @param dhtVer Tx dht version.
+ * @param nearVer Tx near version.
+ * @param msg Local message to log.
+ */
+ public void dumpRemoteTxInfo(UUID nodeId, GridCacheVersion dhtVer, GridCacheVersion nearVer, final String msg) {
+ if (!DIAGNOSTIC_ENABLED)
+ return;
+
+ IgniteInternalFuture<String> fut = diagnosticInfo(nodeId,
+ new IgniteDiagnosticMessage.TxInfoClosure(ctx, dhtVer, nearVer),
+ msg);
+
+ listenAndLog(fut);
+ }
+
+ /**
+ * @param nodeId Target node ID.
+ * @param cacheId Cache ID.
+ * @param keys Keys.
+ * @param msg Local message to log.
+ */
+ public void dumpTxKeyInfo(UUID nodeId, int cacheId, Collection<KeyCacheObject> keys, final String msg) {
+ if (!DIAGNOSTIC_ENABLED)
+ return;
+
+ IgniteInternalFuture<String> fut = diagnosticInfo(nodeId, new IgniteDiagnosticMessage.TxEntriesInfoClosure(ctx, cacheId, keys), msg);
+
+ listenAndLog(fut);
+ }
+
+ /**
+ * @param nodeId Target node ID.
+ * @param msg Local message to log.
+ */
+ public void dumpBasicInfo(final UUID nodeId, final String msg,
+ @Nullable IgniteInClosure<IgniteInternalFuture<String>> lsnr) {
+ if (!DIAGNOSTIC_ENABLED)
+ return;
+
+ IgniteInternalFuture<String> fut = diagnosticInfo(nodeId, new IgniteDiagnosticMessage.BaseClosure(ctx), msg);
+
+ if (lsnr != null)
+ fut.listen(lsnr);
+
+ listenAndLog(fut);
+ }
+
+ /**
+ * @param nodeId Target node ID.
+ * @param topVer Exchange topology version.
+ * @param msg Local message to log.
+ */
+ public void dumpExchangeInfo(final UUID nodeId, AffinityTopologyVersion topVer, final String msg) {
+ if (!DIAGNOSTIC_ENABLED)
+ return;
+
+ IgniteInternalFuture<String> fut = diagnosticInfo(nodeId, new IgniteDiagnosticMessage.ExchangeInfoClosure(ctx, topVer), msg);
+
+ listenAndLog(fut);
+ }
+
+ /**
+ * @param fut Future.
+ */
+ private void listenAndLog(IgniteInternalFuture<String> fut) {
+ fut.listen(new CI1<IgniteInternalFuture<String>>() {
+ @Override public void apply(IgniteInternalFuture<String> fut) {
+ try {
+ diagnosticLog.info(fut.get());
+ }
+ catch (Exception e) {
+ U.error(diagnosticLog, "Failed to dump diagnostic info: " + e, e);
+ }
+ }
+ });
+ }
+
+ /**
+ * @param nodeId Target node ID.
+ * @param c Closure.
+ * @param baseMsg Local message to log.
+ * @return Message future.
+ */
+ private IgniteInternalFuture<String> diagnosticInfo(final UUID nodeId,
+ IgniteClosure<GridKernalContext, String> c,
+ final String baseMsg) {
+ final GridFutureAdapter<String> infoFut = new GridFutureAdapter<>();
+
+ final IgniteInternalFuture<String> rmtFut = sendDiagnosticMessage(nodeId, c);
+
+ rmtFut.listen(new CI1<IgniteInternalFuture<String>>() {
+ @Override public void apply(IgniteInternalFuture<String> fut) {
+ String rmtMsg;
+
+ try {
+ rmtMsg = fut.get();
+ }
+ catch (Exception e) {
+ rmtMsg = "Diagnostic processing error: " + e;
+ }
+
+ final String rmtMsg0 = rmtMsg;
+
+ IgniteInternalFuture<String> locFut = IgniteDiagnosticMessage.dumpCommunicationInfo(ctx, nodeId);
+
+ locFut.listen(new CI1<IgniteInternalFuture<String>>() {
+ @Override public void apply(IgniteInternalFuture<String> locFut) {
+ String locMsg;
+
+ try {
+ locMsg = locFut.get();
+ }
+ catch (Exception e) {
+ locMsg = "Failed to get info for local node: " + e;
+ }
+
+ String msg = baseMsg + U.nl() +
+ "Remote node information:" + U.nl() + rmtMsg0 +
+ U.nl() + "Local communication statistics:" + U.nl() +
+ locMsg;
+
+ infoFut.onDone(msg);
+ }
+ });
+ }
+ });
+
+ return infoFut;
+ }
+
+ /**
+ * @param nodeId Target node ID.
+ * @param c Message closure.
+ * @return Message future.
+ */
+ private IgniteInternalFuture<String> sendDiagnosticMessage(UUID nodeId, IgniteClosure<GridKernalContext, String> c) {
+ try {
+ IgniteDiagnosticMessage msg = IgniteDiagnosticMessage.createRequest(ctx,
+ c,
+ diagFutId.getAndIncrement());
+
+ InternalDiagnosticFuture fut = new InternalDiagnosticFuture(nodeId, msg.futureId());
+
+ diagnosticFuturesMap().put(msg.futureId(), fut);
+
+ ctx.io().sendToGridTopic(nodeId, GridTopic.TOPIC_INTERNAL_DIAGNOSTIC, msg, GridIoPolicy.SYSTEM_POOL);
+
+ return fut;
+ }
+ catch (Exception e) {
+ U.error(log, "Failed to send diagnostic message: " + e);
+
+ return new GridFinishedFuture<>("Failed to send diagnostic message: " + e);
+ }
+ }
+
+ /**
+ * @return Diagnostic messages futures map.
+ */
+ private ConcurrentHashMap<Long, InternalDiagnosticFuture> diagnosticFuturesMap() {
+ ConcurrentHashMap<Long, InternalDiagnosticFuture> map = diagnosticFutMap.get();
+
+ if (map == null) {
+ if (!diagnosticFutMap.compareAndSet(null, map = new ConcurrentHashMap<>()))
+ map = diagnosticFutMap.get();
+ }
+
+ return map;
+ }
+
+ /**
* Update notifier timer task.
*/
private static class UpdateNotifierTimerTask extends GridTimerTask {
@@ -279,4 +587,46 @@ public class ClusterProcessor extends GridProcessorAdapter {
}
}
}
+
+ /**
+ *
+ */
+ class InternalDiagnosticFuture extends GridFutureAdapter<String> {
+ /** */
+ private final long id;
+
+ /** */
+ private final UUID nodeId;
+
+ /**
+ * @param id Future ID.
+ */
+ InternalDiagnosticFuture(UUID nodeId, long id) {
+ this.nodeId = nodeId;
+ this.id = id;
+ }
+
+ /**
+ * @param msg Response message.
+ */
+ public void onResponse(IgniteDiagnosticMessage msg) {
+ onDone(msg.message());
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onDone(@Nullable String res, @Nullable Throwable err) {
+ if (super.onDone(res, err)) {
+ diagnosticFuturesMap().remove(id);
+
+ return true;
+ }
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(InternalDiagnosticFuture.class, this);
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/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 66993ea..b39d9bd 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
@@ -54,7 +54,9 @@ import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.configuration.ConnectorConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.nio.ssl.GridNioSslFilter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.F;
@@ -66,6 +68,7 @@ import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.lang.IgniteReducer;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
@@ -715,6 +718,50 @@ public class GridNioServer<T> {
}
/**
+ * @param msg Message to add.
+ * @param p Session predicate.
+ * @return Future.
+ */
+ public IgniteInternalFuture<String> dumpNodeStats(final String msg, IgnitePredicate<GridNioSession> p) {
+ GridCompoundFuture<String, String> fut = new GridCompoundFuture<>(new IgniteReducer<String, String>() {
+ private final StringBuilder sb = new StringBuilder(msg);
+
+ @Override public boolean collect(@Nullable String msg) {
+ if (!F.isEmpty(msg)) {
+ synchronized (sb) {
+ if (sb.length() > 0)
+ sb.append(U.nl());
+
+ sb.append(msg);
+ }
+ }
+
+ return true;
+ }
+
+ @Override public String reduce() {
+ synchronized (sb) {
+ return sb.toString();
+ }
+ }
+ });
+
+ for (int i = 0; i < clientWorkers.size(); i++) {
+ NioOperationFuture<String> opFut = new NioOperationFuture<>(null, NioOperation.DUMP_STATS);
+
+ opFut.msg = p;
+
+ clientWorkers.get(i).offer(opFut);
+
+ fut.add(opFut);
+ }
+
+ fut.markInitialized();
+
+ return fut;
+ }
+
+ /**
* Establishes a session.
*
* @param ch Channel to register within the server and create session for.
@@ -1812,12 +1859,28 @@ public class GridNioServer<T> {
case DUMP_STATS: {
NioOperationFuture req = (NioOperationFuture)req0;
- try {
- dumpStats();
+ if (req.msg instanceof IgnitePredicate) {
+ StringBuilder sb = new StringBuilder();
+
+ try {
+ dumpStats(sb, (IgnitePredicate<GridNioSession>)req.msg, true);
+ }
+ finally {
+ req.onDone(sb.toString());
+ }
}
- finally {
- // Complete the request just in case (none should wait on this future).
- req.onDone(true);
+ else {
+ try {
+ StringBuilder sb = new StringBuilder();
+
+ dumpStats(sb, null, false);
+
+ U.warn(log, sb.toString());
+ }
+ finally {
+ // Complete the request just in case (none should wait on this future).
+ req.onDone(true);
+ }
}
}
}
@@ -1925,80 +1988,126 @@ public class GridNioServer<T> {
}
/**
- *
+ * @param sb Message builder.
+ * @param keys Keys.
*/
- private void dumpStats() {
- StringBuilder sb = new StringBuilder();
-
- Set<SelectionKey> keys = selector.keys();
-
- sb.append(U.nl())
- .append(">> Selector info [idx=").append(idx)
+ private void dumpSelectorInfo(StringBuilder sb, Set<SelectionKey> keys) {
+ sb.append(">> Selector info [idx=").append(idx)
.append(", keysCnt=").append(keys.size())
.append(", bytesRcvd=").append(bytesRcvd)
.append(", bytesRcvd0=").append(bytesRcvd0)
.append(", bytesSent=").append(bytesSent)
.append(", bytesSent0=").append(bytesSent0)
.append("]").append(U.nl());
+ }
+
+ /**
+ * @param sb Message builder.
+ * @param p Optional session predicate.
+ * @param shortInfo Short info flag.
+ */
+ private void dumpStats(StringBuilder sb,
+ @Nullable IgnitePredicate<GridNioSession> p,
+ boolean shortInfo) {
+ Set<SelectionKey> keys = selector.keys();
+
+ boolean selInfo = p == null;
+
+ if (selInfo)
+ dumpSelectorInfo(sb, keys);
for (SelectionKey key : keys) {
GridSelectorNioSessionImpl ses = (GridSelectorNioSessionImpl)key.attachment();
- MessageWriter writer = ses.meta(MSG_WRITER.ordinal());
- MessageReader reader = ses.meta(GridDirectParser.READER_META_KEY);
+ boolean sesInfo = p == null || p.apply(ses);
- sb.append(" Connection info [")
- .append("in=").append(ses.accepted())
- .append(", rmtAddr=").append(ses.remoteAddress())
- .append(", locAddr=").append(ses.localAddress());
+ if (sesInfo) {
+ if (!selInfo) {
+ dumpSelectorInfo(sb, keys);
- GridNioRecoveryDescriptor outDesc = ses.outRecoveryDescriptor();
+ selInfo = true;
+ }
- if (outDesc != null) {
- sb.append(", msgsSent=").append(outDesc.sent())
- .append(", msgsAckedByRmt=").append(outDesc.acked())
- .append(", descIdHash=").append(System.identityHashCode(outDesc));
- }
- else
- sb.append(", outRecoveryDesc=null");
+ sb.append(" Connection info [")
+ .append("in=").append(ses.accepted())
+ .append(", rmtAddr=").append(ses.remoteAddress())
+ .append(", locAddr=").append(ses.localAddress());
- GridNioRecoveryDescriptor inDesc = ses.inRecoveryDescriptor();
+ GridNioRecoveryDescriptor outDesc = ses.outRecoveryDescriptor();
- if (inDesc != null) {
- sb.append(", msgsRcvd=").append(inDesc.received())
- .append(", lastAcked=").append(inDesc.lastAcknowledged())
- .append(", descIdHash=").append(System.identityHashCode(inDesc));
- }
- else
- sb.append(", inRecoveryDesc=null");
+ if (outDesc != null) {
+ sb.append(", msgsSent=").append(outDesc.sent())
+ .append(", msgsAckedByRmt=").append(outDesc.acked())
+ .append(", descIdHash=").append(System.identityHashCode(outDesc));
+
+ if (!outDesc.messagesRequests().isEmpty()) {
+ int cnt = 0;
+
+ sb.append(", unackedMsgs=[");
+
+ for (SessionWriteRequest req : outDesc.messagesRequests()) {
+ if (cnt != 0)
+ sb.append(", ");
- sb.append(", bytesRcvd=").append(ses.bytesReceived())
- .append(", bytesRcvd0=").append(ses.bytesReceived0())
- .append(", bytesSent=").append(ses.bytesSent())
- .append(", bytesSent0=").append(ses.bytesSent0())
- .append(", opQueueSize=").append(ses.writeQueueSize())
- .append(", msgWriter=").append(writer != null ? writer.toString() : "null")
- .append(", msgReader=").append(reader != null ? reader.toString() : "null");
+ Object msg = req.message();
- int cnt = 0;
+ if (shortInfo && msg instanceof GridIoMessage)
+ msg = ((GridIoMessage)msg).message().getClass().getSimpleName();
- for (SessionWriteRequest req : ses.writeQueue()) {
- if (cnt == 0)
- sb.append(",\n opQueue=[").append(req);
+ sb.append(msg);
+
+ if (++cnt == 5)
+ break;
+ }
+
+ sb.append(']');
+ }
+ }
else
- sb.append(',').append(req);
+ sb.append(", outRecoveryDesc=null");
- if (++cnt == 5) {
- sb.append(']');
+ GridNioRecoveryDescriptor inDesc = ses.inRecoveryDescriptor();
- break;
+ if (inDesc != null) {
+ sb.append(", msgsRcvd=").append(inDesc.received())
+ .append(", lastAcked=").append(inDesc.lastAcknowledged())
+ .append(", descIdHash=").append(System.identityHashCode(inDesc));
}
- }
+ else
+ sb.append(", inRecoveryDesc=null");
- sb.append("]").append(U.nl());
- }
+ sb.append(", bytesRcvd=").append(ses.bytesReceived())
+ .append(", bytesRcvd0=").append(ses.bytesReceived0())
+ .append(", bytesSent=").append(ses.bytesSent())
+ .append(", bytesSent0=").append(ses.bytesSent0())
+ .append(", opQueueSize=").append(ses.writeQueueSize());
+
+ if (!shortInfo) {
+ MessageWriter writer = ses.meta(MSG_WRITER.ordinal());
+ MessageReader reader = ses.meta(GridDirectParser.READER_META_KEY);
+
+ sb.append(", msgWriter=").append(writer != null ? writer.toString() : "null")
+ .append(", msgReader=").append(reader != null ? reader.toString() : "null");
+ }
+
+ int cnt = 0;
- U.warn(log, sb.toString());
+ for (SessionWriteRequest req : ses.writeQueue()) {
+ if (cnt == 0)
+ sb.append(",\n opQueue=[").append(req);
+ else
+ sb.append(',').append(req);
+
+ if (++cnt == 5) {
+ sb.append(']');
+
+ break;
+ }
+ }
+
+ sb.append("]");
+ }
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/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 2e35c6e..3670fb4 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
@@ -67,6 +67,7 @@ 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;
@@ -1764,68 +1765,46 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
}
/**
- * Dumps SPI per-connection stats to logs.
+ * @param nodeId Target node ID.
+ * @return Future.
*/
- public void dumpStats() {
- IgniteLogger log = this.log;
+ public IgniteInternalFuture<String> dumpNodeStatistics(final UUID nodeId) {
+ StringBuilder sb = new StringBuilder("Communication SPI statistics [rmtNode=").append(nodeId).append(']').append(U.nl());
- if (log != null) {
- StringBuilder sb = new StringBuilder("Communication SPI recovery descriptors: ").append(U.nl());
-
- for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : recoveryDescs.entrySet()) {
- GridNioRecoveryDescriptor desc = entry.getValue();
-
- sb.append(" [key=").append(entry.getKey())
- .append(", msgsSent=").append(desc.sent())
- .append(", msgsAckedByRmt=").append(desc.acked())
- .append(", msgsRcvd=").append(desc.received())
- .append(", lastAcked=").append(desc.lastAcknowledged())
- .append(", reserveCnt=").append(desc.reserveCount())
- .append(", descIdHash=").append(System.identityHashCode(desc))
- .append(']').append(U.nl());
- }
-
- for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : outRecDescs.entrySet()) {
- GridNioRecoveryDescriptor desc = entry.getValue();
-
- sb.append(" [key=").append(entry.getKey())
- .append(", msgsSent=").append(desc.sent())
- .append(", msgsAckedByRmt=").append(desc.acked())
- .append(", reserveCnt=").append(desc.reserveCount())
- .append(", connected=").append(desc.connected())
- .append(", reserved=").append(desc.reserved())
- .append(", descIdHash=").append(System.identityHashCode(desc))
- .append(']').append(U.nl());
- }
+ dumpInfo(sb, nodeId);
- for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : inRecDescs.entrySet()) {
- GridNioRecoveryDescriptor desc = entry.getValue();
-
- sb.append(" [key=").append(entry.getKey())
- .append(", msgsRcvd=").append(desc.received())
- .append(", lastAcked=").append(desc.lastAcknowledged())
- .append(", reserveCnt=").append(desc.reserveCount())
- .append(", connected=").append(desc.connected())
- .append(", reserved=").append(desc.reserved())
- .append(", handshakeIdx=").append(desc.handshakeIndex())
- .append(", descIdHash=").append(System.identityHashCode(desc))
- .append(']').append(U.nl());
- }
+ GridNioServer<Message> nioSrvr = this.nioSrvr;
- sb.append("Communication SPI clients: ").append(U.nl());
+ if (nioSrvr != null) {
+ sb.append("NIO sessions statistics:");
- for (Map.Entry<UUID, GridCommunicationClient[]> entry : clients.entrySet()) {
- UUID nodeId = entry.getKey();
- GridCommunicationClient[] clients0 = entry.getValue();
+ IgnitePredicate<GridNioSession> p = new IgnitePredicate<GridNioSession>() {
+ @Override public boolean apply(GridNioSession ses) {
+ ConnectionKey connId = ses.meta(CONN_IDX_META);
- for (GridCommunicationClient client : clients0) {
- if (client != null) {
- sb.append(" [node=").append(nodeId)
- .append(", client=").append(client)
- .append(']').append(U.nl());
- }
+ return connId != null && nodeId.equals(connId.nodeId());
}
- }
+ };
+
+ return nioSrvr.dumpNodeStats(sb.toString(), p);
+ }
+ else {
+ sb.append(U.nl()).append("GridNioServer is null.");
+
+ return new GridFinishedFuture<>(sb.toString());
+ }
+ }
+
+ /**
+ * Dumps SPI per-connection stats to logs.
+ */
+ public void dumpStats() {
+ IgniteLogger log = this.log;
+
+ if (log != null) {
+ StringBuilder sb = new StringBuilder();
+
+ dumpInfo(sb, null);
U.warn(log, sb.toString());
}
@@ -1836,11 +1815,81 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
nioSrvr.dumpStats();
}
- /** */
- private final ThreadLocal<Integer> threadConnIdx = new ThreadLocal<>();
+ /**
+ * @param sb Message builder.
+ * @param dstNodeId Target node ID.
+ */
+ private void dumpInfo(StringBuilder sb, UUID dstNodeId) {
+ sb.append("Communication SPI recovery descriptors: ").append(U.nl());
- /** */
- private final AtomicInteger connIdx = new AtomicInteger();
+ for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : recoveryDescs.entrySet()) {
+ GridNioRecoveryDescriptor desc = entry.getValue();
+
+ if (dstNodeId != null && !dstNodeId.equals(entry.getKey().nodeId()))
+ continue;
+
+ sb.append(" [key=").append(entry.getKey())
+ .append(", msgsSent=").append(desc.sent())
+ .append(", msgsAckedByRmt=").append(desc.acked())
+ .append(", msgsRcvd=").append(desc.received())
+ .append(", lastAcked=").append(desc.lastAcknowledged())
+ .append(", reserveCnt=").append(desc.reserveCount())
+ .append(", descIdHash=").append(System.identityHashCode(desc))
+ .append(']').append(U.nl());
+ }
+
+ for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : outRecDescs.entrySet()) {
+ GridNioRecoveryDescriptor desc = entry.getValue();
+
+ if (dstNodeId != null && !dstNodeId.equals(entry.getKey().nodeId()))
+ continue;
+
+ sb.append(" [key=").append(entry.getKey())
+ .append(", msgsSent=").append(desc.sent())
+ .append(", msgsAckedByRmt=").append(desc.acked())
+ .append(", reserveCnt=").append(desc.reserveCount())
+ .append(", connected=").append(desc.connected())
+ .append(", reserved=").append(desc.reserved())
+ .append(", descIdHash=").append(System.identityHashCode(desc))
+ .append(']').append(U.nl());
+ }
+
+ for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : inRecDescs.entrySet()) {
+ GridNioRecoveryDescriptor desc = entry.getValue();
+
+ if (dstNodeId != null && !dstNodeId.equals(entry.getKey().nodeId()))
+ continue;
+
+ sb.append(" [key=").append(entry.getKey())
+ .append(", msgsRcvd=").append(desc.received())
+ .append(", lastAcked=").append(desc.lastAcknowledged())
+ .append(", reserveCnt=").append(desc.reserveCount())
+ .append(", connected=").append(desc.connected())
+ .append(", reserved=").append(desc.reserved())
+ .append(", handshakeIdx=").append(desc.handshakeIndex())
+ .append(", descIdHash=").append(System.identityHashCode(desc))
+ .append(']').append(U.nl());
+ }
+
+ sb.append("Communication SPI clients: ").append(U.nl());
+
+ for (Map.Entry<UUID, GridCommunicationClient[]> entry : clients.entrySet()) {
+ UUID clientNodeId = entry.getKey();
+
+ if (dstNodeId != null && !dstNodeId.equals(clientNodeId))
+ continue;
+
+ GridCommunicationClient[] clients0 = entry.getValue();
+
+ for (GridCommunicationClient client : clients0) {
+ if (client != null) {
+ sb.append(" [node=").append(clientNodeId)
+ .append(", client=").append(client)
+ .append(']').append(U.nl());
+ }
+ }
+ }
+ }
/** {@inheritDoc} */
@Override public Map<String, Object> getNodeAttributes() throws IgniteSpiException {
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesMultipleConnectionsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesMultipleConnectionsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesMultipleConnectionsTest.java
new file mode 100644
index 0000000..6b77704
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesMultipleConnectionsTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.managers;
+
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+
+/**
+ *
+ */
+public class IgniteDiagnosticMessagesMultipleConnectionsTest extends IgniteDiagnosticMessagesTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setConnectionsPerNode(5);
+
+ return cfg;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/e3dd573d/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesTest.java
new file mode 100644
index 0000000..95f7513
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/IgniteDiagnosticMessagesTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.managers;
+
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.lang.IgniteInClosure;
+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.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ *
+ */
+public class IgniteDiagnosticMessagesTest extends GridCommonAbstractTest {
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private boolean client;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setClientMode(client);
+
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setCacheMode(CacheMode.REPLICATED);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setName("c1");
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ System.setProperty(IgniteSystemProperties.IGNITE_DIAGNOSTIC_ENABLED, "true");
+
+ startGrids(3);
+
+ client = true;
+
+ startGrid(3);
+
+ startGrid(4);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ System.setProperty(IgniteSystemProperties.IGNITE_DIAGNOSTIC_ENABLED, "false");
+
+ stopAllGrids();
+
+ super.afterTestsStopped();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDiagnosticMessages() throws Exception {
+ awaitPartitionMapExchange();
+
+ sendDiagnostic();
+
+ for (int i = 0; i < 5; i++) {
+ final IgniteCache cache = ignite(i).cache("c1");
+
+ GridTestUtils.runMultiThreaded(new Runnable() {
+ @Override public void run() {
+ for (int j = 0; j < 10; j++)
+ cache.put(ThreadLocalRandom.current().nextInt(), j);
+ }
+ }, 10, "cache-thread");
+ }
+
+ sendDiagnostic();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void sendDiagnostic() throws Exception {
+ for (int i = 0; i < 5; i++) {
+ IgniteKernal node = (IgniteKernal)ignite(i);
+
+ for (int j = 0; j < 5; j++) {
+ if (i != j) {
+ ClusterNode dstNode = ignite(j).cluster().localNode();
+
+ final GridFutureAdapter<String> fut = new GridFutureAdapter<>();
+
+ node.context().cluster().dumpBasicInfo(dstNode.id(), "Test diagnostic",
+ new IgniteInClosure<IgniteInternalFuture<String>>() {
+ @Override public void apply(IgniteInternalFuture<String> diagFut) {
+ try {
+ fut.onDone(diagFut.get());
+ }
+ catch (Exception e) {
+ fut.onDone(e);
+ }
+ }
+ }
+ );
+
+ String msg = fut.get();
+
+ assertTrue("Unexpected message: " + msg,
+ msg.contains("Test diagnostic") &&
+ msg.contains("General node info [id=" + dstNode.id() + ", client=" + dstNode.isClient() + ", discoTopVer=AffinityTopologyVersion [topVer=5, minorTopVer=") &&
+ msg.contains("Partitions exchange info [readyVer=AffinityTopologyVersion [topVer=5, minorTopVer="));
+ }
+ }
+ }
+ }
+}