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/03/07 10:33:58 UTC
[4/4] ignite git commit: ignite-4705
ignite-4705
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b1992a92
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b1992a92
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b1992a92
Branch: refs/heads/ignite-4705-debug
Commit: b1992a92ba953080f13d01d4c2c72c898373978f
Parents: 17cf339
Author: sboikov <sb...@gridgain.com>
Authored: Tue Mar 7 13:33:45 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue Mar 7 13:33:45 2017 +0300
----------------------------------------------------------------------
.../GridCachePartitionExchangeManager.java | 10 +
.../GridNearAtomicSingleUpdateFuture.java | 4 +
.../distributed/dht/atomic/TestDebugLog.java | 394 +++++++++++++++++++
.../cache/GridCachePutAllFailoverSelfTest.java | 7 +
.../processors/cache/GridCachePutAllTask.java | 6 +-
...IgniteCacheAtomicPutAllFailoverSelfTest.java | 4 +
.../IgniteCacheRestartTestSuite2.java | 10 -
7 files changed, 423 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/b1992a92/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 14a0fb4..e7bfdd4 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
@@ -64,6 +64,8 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartit
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicAbstractUpdateFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicSingleUpdateFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.TestDebugLog;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
@@ -1424,6 +1426,14 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
if (longRunningOpsDumpCnt < GridDhtPartitionsExchangeFuture.DUMP_PENDING_OBJECTS_THRESHOLD) {
U.warn(log, "Found long running cache future [startTime=" + formatTime(fut.startTime()) +
", curTime=" + formatTime(curTime) + ", fut=" + fut + ']');
+
+ if (fut instanceof GridNearAtomicSingleUpdateFuture) {
+ TestDebugLog.addEntryMessage(((GridNearAtomicSingleUpdateFuture) fut).key(), null, "hang fut");
+
+ TestDebugLog.printKeyMessages("test_debug.txt", ((GridNearAtomicSingleUpdateFuture) fut).key());
+
+ System.exit(44);
+ }
}
else
break;
http://git-wip-us.apache.org/repos/asf/ignite/blob/b1992a92/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
index a113b27..fccb23d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
@@ -124,6 +124,10 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
this.val = val;
}
+ public Object key() {
+ return key;
+ }
+
/** {@inheritDoc} */
@Override public Long id() {
synchronized (mux) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/b1992a92/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/TestDebugLog.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/TestDebugLog.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/TestDebugLog.java
new file mode 100644
index 0000000..bf4f4f0
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/TestDebugLog.java
@@ -0,0 +1,394 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
+
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCustomEventMessage;
+
+/**
+ * TODO
+ if (!cacheMsg.partitionExchangeMessage() && !(cacheMsg instanceof GridDhtPartitionDemandMessage) && !(cacheMsg instanceof GridDhtPartitionSupplyMessage))
+ TestDebugLog.addMessage("Message: " + cacheMsg);
+ */
+public class TestDebugLog {
+ /** */
+ private static final List<Object> msgs = Collections.synchronizedList(new ArrayList<>(1_000_000));
+
+ /** */
+ private static final SimpleDateFormat DEBUG_DATE_FMT = new SimpleDateFormat("HH:mm:ss,SSS");
+
+ private static final boolean disabled = false;
+
+ static class Message {
+ String thread = Thread.currentThread().getName();
+
+ String msg;
+
+ Object val;
+
+ long ts = U.currentTimeMillis();
+
+ public Message(String msg) {
+ this.msg = msg;
+ }
+
+ public Message(String msg, Object val) {
+ this.msg = msg;
+ this.val = val;
+ }
+
+ public String toString() {
+ return "Msg [thread=" + thread + ", time=" + DEBUG_DATE_FMT.format(new Date(ts)) +
+ ", msg=" + msg + ", val=" + val + ']';
+ }
+ }
+
+ static class TxMessage extends Message {
+ Object ver;
+ Object val;
+
+ public TxMessage(Object ver, Object val, String msg) {
+ super(msg);
+
+ this.ver = ver;
+ this.val = val;
+ }
+
+ public String toString() {
+ String s = "TxMsg [ver=" + ver +
+ ", msg=" + msg +
+ ", thread=" + thread +
+ ", val=" + val +
+ ", time=" + DEBUG_DATE_FMT.format(new Date(ts)) + ']';
+
+// if (val instanceof Exception) {
+// System.out.println("Error for " + s);
+//
+// ((Exception) val).printStackTrace(System.out);
+// }
+
+ return s;
+ }
+ }
+
+ static class PartMessage extends Message {
+ private final int cacheId;
+ private final int partId;
+ private Object val;
+
+ public PartMessage(int cacheId, int partId, Object val, String msg) {
+ super(msg);
+
+ this.cacheId = cacheId;
+ this.partId = partId;
+ this.val = val;
+ this.msg = msg;
+ }
+
+ boolean match(int cacheId, int partId) {
+ return this.cacheId == cacheId && this.partId == partId;
+ }
+
+ @Override public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ PartMessage partKey = (PartMessage) o;
+
+ if (cacheId != partKey.cacheId) return false;
+ return partId == partKey.partId;
+
+ }
+
+ @Override public int hashCode() {
+ int result = cacheId;
+ result = 31 * result + partId;
+ return result;
+ }
+
+ public String toString() {
+ return "PartMessage [partId=" + partId +
+ ", val=" + val +
+ ", msg=" + msg +
+ ", thread=" + thread +
+ ", time=" + DEBUG_DATE_FMT.format(new Date(ts)) +
+ ", cacheId=" + cacheId + ']';
+ }
+ }
+
+ static class EntryMessage extends Message {
+ Object key;
+ Object val;
+
+ public EntryMessage(Object key, Object val, String msg) {
+ super(msg);
+
+ this.key = key;
+ this.val = val;
+ }
+
+ public String toString() {
+ //return "EntryMsg [key=" + key + ", msg=" + msg + ", thread=" + thread + ", time=" + DEBUG_DATE_FMT.format(new Date(ts)) + ", val=" + val + ']';
+
+ String s = "EntryMsg [key=" + key +
+ ", thread=" + thread +
+ ", msg=" + msg +
+ ", val=" + val +
+ ", time=" + DEBUG_DATE_FMT.format(new Date(ts)) + ']';
+
+// if (val instanceof Exception) {
+// System.out.println("Error for " + s);
+//
+// ((Exception) val).printStackTrace(System.out);
+// }
+
+ return s;
+ }
+ }
+
+ static final boolean out = false;
+
+ public static void startRoutineMsg(TcpDiscoveryAbstractMessage msg, Marshaller marsh, String txt) {
+ if (msg instanceof TcpDiscoveryCustomEventMessage)
+ startRoutineMsg((TcpDiscoveryCustomEventMessage)msg, marsh, txt);
+ }
+
+ public static void startRoutineMsg(TcpDiscoveryCustomEventMessage msg, Marshaller marsh, String txt) {
+// try {
+// CustomMessageWrapper msg0 = (CustomMessageWrapper)msg.message(marsh);
+//
+// DiscoveryCustomMessage custMsg = msg0.delegate();
+//
+// if (custMsg instanceof StartRoutineDiscoveryMessage) {
+// StartRoutineDiscoveryMessage startMsg = (StartRoutineDiscoveryMessage)custMsg;
+//
+// addEntryMessage(startMsg.routineId(), "start msg", txt);
+// }
+// else if (custMsg instanceof StartRoutineAckDiscoveryMessage) {
+// StartRoutineAckDiscoveryMessage startMsg = (StartRoutineAckDiscoveryMessage)custMsg;
+//
+// addEntryMessage(startMsg.routineId(), "start ack msg", txt);
+// }
+// }
+// catch (Throwable e) {
+// e.printStackTrace();
+//
+// System.exit(55);
+// }
+ }
+
+ public static void addMessage(String msg) {
+// if (disabled)
+// return;
+//
+ msgs.add(new Message(msg));
+
+ if (out)
+ System.out.println(msg);
+ }
+
+ public static void addMessage(String msg, Object val) {
+ if (disabled)
+ return;
+
+ msgs.add(new Message(msg, val));
+
+ if (out)
+ System.out.println(msg);
+ }
+
+ public static void addKeyMessage(CacheObject key, Object val, String msg) {
+ addEntryMessage(key.value(null, false), val, msg);
+ }
+
+ public static void addQueueItemKeyMessage(KeyCacheObject key, Object val, String msg) {
+// Object q = key.value(null, false);
+//
+// if (q instanceof GridCacheQueueItemKey)
+// addEntryMessage(((GridCacheQueueItemKey)q).index(), val, msg);
+// else
+// addEntryMessage(q, val, msg);
+ }
+
+ public static void addEntryMessage(Object key, Object val, String msg) {
+ if (disabled)
+ return;
+
+ assert key != null;
+
+ if (key instanceof CacheObject)
+ key = ((CacheObject) key).value(null, false);
+
+ EntryMessage msg0 = new EntryMessage(key, val, msg);
+
+ msgs.add(msg0);
+
+ if (out)
+ System.out.println(msg0.toString());
+ }
+
+ public static void printMessages(String fileName) {
+ List<Object> msgs0;
+
+ synchronized (msgs) {
+ msgs0 = new ArrayList<>(msgs);
+
+ // msgs.clear();
+ }
+
+ if (fileName != null) {
+ try {
+ FileOutputStream out = new FileOutputStream(fileName);
+
+ PrintWriter w = new PrintWriter(out);
+
+ for (Object msg : msgs0)
+ w.println(msg.toString());
+
+ w.close();
+
+ out.close();
+ }
+ catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ else {
+ for (Object msg : msgs0)
+ System.out.println(msg);
+ }
+ }
+
+ public static void addPartMessage(int cacheId,
+ int partId,
+ Object val,
+ String msg) {
+ msgs.add(new PartMessage(cacheId, partId, val, msg));
+ }
+
+ public static void printKeyMessages(String fileName, Object key) {
+ assert key != null;
+
+ List<Object> msgs0;
+
+ synchronized (msgs) {
+ msgs0 = new ArrayList<>(msgs);
+
+ //msgs.clear();
+ }
+
+ if (fileName != null) {
+ try {
+ FileOutputStream out = new FileOutputStream(fileName);
+
+ PrintWriter w = new PrintWriter(out);
+
+ for (Object msg : msgs0) {
+ if (msg instanceof EntryMessage && !((EntryMessage)msg).key.equals(key))
+ continue;
+
+ w.println(msg.toString());
+ }
+
+ w.close();
+
+ out.close();
+ }
+ catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ else {
+ for (Object msg : msgs0) {
+ if (msg instanceof EntryMessage && !((EntryMessage)msg).key.equals(key))
+ continue;
+
+ System.out.println(msg);
+ }
+ }
+ }
+ public static void printKeyAndPartMessages(String fileName, Object key, int partId, int cacheId) {
+ //assert key != null;
+
+ List<Object> msgs0;
+
+ synchronized (msgs) {
+ msgs0 = new ArrayList<>(msgs);
+
+ //msgs.clear();
+ }
+
+ if (fileName != null) {
+ try {
+ FileOutputStream out = new FileOutputStream(fileName);
+
+ PrintWriter w = new PrintWriter(out);
+
+ for (Object msg : msgs0) {
+ if (key != null) {
+ if (msg instanceof EntryMessage && !((EntryMessage)msg).key.equals(key))
+ continue;
+ }
+
+ if (msg instanceof PartMessage) {
+ PartMessage pm = (PartMessage)msg;
+
+ if (pm.cacheId != cacheId || pm.partId != partId)
+ continue;
+ }
+
+ w.println(msg.toString());
+ }
+
+ w.close();
+
+ out.close();
+ }
+ catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ else {
+// for (Object msg : msgs0) {
+// if (msg instanceof EntryMessage && !((EntryMessage)msg).key.equals(key))
+// continue;
+//
+// System.out.println(msg);
+// }
+ }
+ }
+
+ public static void clear() {
+ msgs.clear();
+ }
+
+ public static void main(String[] args) {
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b1992a92/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllFailoverSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllFailoverSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllFailoverSelfTest.java
index b48bbc7..9c95123 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllFailoverSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllFailoverSelfTest.java
@@ -46,6 +46,7 @@ import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.TestDebugLog;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.PN;
@@ -436,6 +437,12 @@ public class GridCachePutAllFailoverSelfTest extends GridCommonAbstractTest {
if (entry0 != null)
info(" >>> " + entry);
}
+
+ TestDebugLog.addEntryMessage(key, null, "absent key");
+
+ TestDebugLog.printKeyMessages("test_debug.txt", key);
+
+ System.exit(55);
}
info("");
http://git-wip-us.apache.org/repos/asf/ignite/blob/b1992a92/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllTask.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllTask.java
index e728a0f..3c4b67f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllTask.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllTask.java
@@ -120,7 +120,8 @@ class GridCachePutAllTask extends ComputeTaskAdapter<Collection<Integer>, Void>
for (int i = 0; i < RETRIES; i++) {
try {
- cache.putAll(putMap);
+ for (Map.Entry<Integer, Integer> e : putMap.entrySet())
+ cache.put(e.getKey(), e.getValue());
break;
}
@@ -148,7 +149,8 @@ class GridCachePutAllTask extends ComputeTaskAdapter<Collection<Integer>, Void>
for (int i = 0; i < RETRIES; i++) {
try {
- cache.putAll(putMap);
+ for (Map.Entry<Integer, Integer> e : putMap.entrySet())
+ cache.put(e.getKey(), e.getValue());
break;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b1992a92/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAtomicPutAllFailoverSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAtomicPutAllFailoverSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAtomicPutAllFailoverSelfTest.java
index 3f9fc5c..ded4e73 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAtomicPutAllFailoverSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAtomicPutAllFailoverSelfTest.java
@@ -34,4 +34,8 @@ public class IgniteCacheAtomicPutAllFailoverSelfTest extends GridCachePutAllFail
@Override public void testPutAllFailoverColocatedNearEnabledTwoBackupsOffheapTieredSwap(){
fail("https://issues.apache.org/jira/browse/IGNITE-1584");
}
+
+ public void testPutAllFailoverNearDisabledThreeBackups() throws Exception {
+ super.testPutAllFailoverNearDisabledThreeBackups();
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/b1992a92/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
index 4ac2cde..11f0428 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
@@ -37,17 +37,7 @@ public class IgniteCacheRestartTestSuite2 extends TestSuite {
public static TestSuite suite() throws Exception {
TestSuite suite = new TestSuite("Cache Restart Test Suite2");
- suite.addTestSuite(IgniteCacheAtomicNodeRestartTest.class);
- suite.addTestSuite(IgniteCacheAtomicReplicatedNodeRestartSelfTest.class);
-
suite.addTestSuite(IgniteCacheAtomicPutAllFailoverSelfTest.class);
- suite.addTestSuite(IgniteCachePutAllRestartTest.class);
- suite.addTestSuite(GridCachePutAllFailoverSelfTest.class);
-
- // TODO IGNITE-4705.
- //suite.addTestSuite(IgniteBinaryMetadataUpdateNodeRestartTest.class);
-
- suite.addTestSuite(IgniteCacheGetRestartTest.class);
return suite;
}