You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ra...@apache.org on 2015/11/11 01:10:36 UTC
[21/50] [abbrv] ignite git commit: ignite-1843 Avoid discovery thread
blocking in GridJobProcessor.
ignite-1843 Avoid discovery thread blocking in GridJobProcessor.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/705ee1ee
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/705ee1ee
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/705ee1ee
Branch: refs/heads/ignite-1527
Commit: 705ee1ee6a3b373d5af5d89a27fbf8d1a2fcb513
Parents: 740c1ed
Author: sboikov <sb...@gridgain.com>
Authored: Wed Nov 4 12:52:20 2015 +0300
Committer: Raul Kripalani <ra...@apache.org>
Committed: Wed Nov 11 00:09:42 2015 +0000
----------------------------------------------------------------------
.../apache/ignite/internal/IgniteKernal.java | 1 +
.../affinity/GridAffinityAssignmentCache.java | 13 +++
.../cache/GridCacheAffinityManager.java | 10 ++
.../GridCachePartitionExchangeManager.java | 24 +++++
.../processors/cache/GridCachePreloader.java | 5 +
.../cache/GridCachePreloaderAdapter.java | 5 +
.../dht/preloader/GridDhtForceKeysFuture.java | 14 +++
.../dht/preloader/GridDhtPreloader.java | 37 ++++++++
.../processors/job/GridJobProcessor.java | 96 +++++++++-----------
.../GridTaskFailoverAffinityRunTest.java | 3 +
...niteCacheClientNodeChangingTopologyTest.java | 4 +-
11 files changed, 158 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/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 0277acc..026f8ca 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
@@ -3170,6 +3170,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
U.warn(log, "Dumping debug info for node [id=" + locNode.id() +
", name=" + ctx.gridName() +
", order=" + locNode.order() +
+ ", topVer=" + ctx.discovery().topologyVersion() +
", client=" + client +
(client && routerId != null ? ", routerId=" + routerId : "") + ']');
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
index 18776a4..8bc40cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
@@ -41,6 +41,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheDefaultAffinityKeyMa
import org.apache.ignite.internal.processors.cache.GridCacheInternal;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
import org.jsr166.ConcurrentLinkedHashMap;
@@ -409,6 +410,18 @@ public class GridAffinityAssignmentCache {
}
/**
+ * Dumps debug information.
+ */
+ public void dumpDebugInfo() {
+ if (!readyFuts.isEmpty()) {
+ U.warn(log, "Pending affinity ready futures [cache=" + cacheName + "]:");
+
+ for (AffinityReadyFuture fut : readyFuts.values())
+ U.warn(log, ">>> " + fut);
+ }
+ }
+
+ /**
* Get cached affinity for specified topology version.
*
* @param topVer Topology version.
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
index 5c43205..eddffea 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
@@ -384,4 +384,14 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter {
public AffinityTopologyVersion affinityTopologyVersion() {
return aff.lastVersion();
}
+
+ /**
+ * Dumps debug information.
+ */
+ public void dumpDebugInfo() {
+ GridAffinityAssignmentCache aff0 = aff;
+
+ if (aff0 != null)
+ aff0.dumpDebugInfo();
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/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 adc2174..c8ee6e3 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
@@ -992,6 +992,13 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
for (GridDhtPartitionsExchangeFuture fut : exchWorker.futQ)
U.warn(log, ">>> " + fut);
+ if (!readyFuts.isEmpty()) {
+ U.warn(log, "Pending affinity ready futures:");
+
+ for (AffinityReadyFuture fut : readyFuts.values())
+ U.warn(log, ">>> " + fut);
+ }
+
ExchangeFutureSet exchFuts = this.exchFuts;
if (exchFuts != null) {
@@ -1041,6 +1048,23 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
for (GridCacheFuture<?> fut : mvcc.atomicFutures())
U.warn(log, ">>> " + fut);
}
+
+ for (GridCacheContext ctx : cctx.cacheContexts()) {
+ if (ctx.isLocal())
+ continue;
+
+ GridCacheContext ctx0 = ctx.isNear() ? ctx.near().dht().context() : ctx;
+
+ GridCachePreloader preloader = ctx0.preloader();
+
+ if (preloader != null)
+ preloader.dumpDebugInfo();
+
+ GridCacheAffinityManager affMgr = ctx0.affinity();
+
+ if (affMgr != null)
+ affMgr.dumpDebugInfo();
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
index 755958e..1edaef2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
@@ -132,4 +132,9 @@ public interface GridCachePreloader {
* Unwinds undeploys.
*/
public void unwindUndeploys();
+
+ /**
+ * Dumps debug information.
+ */
+ public void dumpDebugInfo();
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloaderAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloaderAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloaderAdapter.java
index 5405449..4ec6749 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloaderAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloaderAdapter.java
@@ -146,4 +146,9 @@ public class GridCachePreloaderAdapter implements GridCachePreloader {
@Override public void addAssignments(GridDhtPreloaderAssignments assignments, boolean forcePreload) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public void dumpDebugInfo() {
+ // No-op.
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
index bb78748..db0e780 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.GridLeanSet;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -394,6 +395,19 @@ public final class GridDhtForceKeysFuture<K, V> extends GridCompoundFuture<Objec
return mappings;
}
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
+ @Override public String apply(IgniteInternalFuture<?> f) {
+ return f.toString();
+ }
+ });
+
+ return S.toString(GridDhtForceKeysFuture.class, this,
+ "innerFuts", futs,
+ "super", super.toString());
+ }
+
/**
* Mini-future for get operations. Mini-futures are only waiting on a single
* node as opposed to multiple nodes.
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
index 356a85b..fe85968 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
@@ -96,6 +96,9 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
private ConcurrentMap<AffinityTopologyVersion, GridDhtAssignmentFetchFuture> pendingAssignmentFetchFuts =
new ConcurrentHashMap8<>();
+ /** Stop flag. */
+ private volatile boolean stopping;
+
/** Discovery listener. */
private final GridLocalEventListener discoLsnr = new GridLocalEventListener() {
@Override public void onEvent(Event evt) {
@@ -218,6 +221,8 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
if (log.isDebugEnabled())
log.debug("DHT rebalancer onKernalStop callback.");
+ stopping = true;
+
cctx.events().removeListener(discoLsnr);
// Acquire write busy lock.
@@ -229,6 +234,11 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
if (demandPool != null)
demandPool.stop();
+ IgniteCheckedException err = stopError();
+
+ for (GridDhtForceKeysFuture fut : forceKeyFuts.values())
+ fut.onDone(err);
+
top = null;
}
@@ -595,6 +605,9 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
*/
void addFuture(GridDhtForceKeysFuture<?, ?> fut) {
forceKeyFuts.put(fut.futureId(), fut);
+
+ if (stopping)
+ fut.onDone(stopError());
}
/**
@@ -607,6 +620,30 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
}
/**
+ * @return Node stop exception.
+ */
+ private IgniteCheckedException stopError() {
+ return new IgniteCheckedException("Operation has been cancelled (cache or node is stopping).");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void dumpDebugInfo() {
+ if (!forceKeyFuts.isEmpty()) {
+ U.warn(log, "Pending force key futures [cache=" + cctx.name() +"]:");
+
+ for (GridDhtForceKeysFuture fut : forceKeyFuts.values())
+ U.warn(log, ">>> " + fut);
+ }
+
+ if (!pendingAssignmentFetchFuts.isEmpty()) {
+ U.warn(log, "Pending assignment fetch futures [cache=" + cctx.name() +"]:");
+
+ for (GridDhtAssignmentFetchFuture fut : pendingAssignmentFetchFuts.values())
+ U.warn(log, ">>> " + fut);
+ }
+ }
+
+ /**
*
*/
private abstract class MessageHandler<M> implements IgniteBiInClosure<UUID, M> {
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
index 4d6d0bf..20bf58c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
@@ -202,7 +202,7 @@ public class GridJobProcessor extends GridProcessorAdapter {
};
/** Current session. */
- private final ThreadLocal<ComputeTaskSession> currentSess = new ThreadLocal<>();
+ private final ThreadLocal<ComputeTaskSession> currSess = new ThreadLocal<>();
/**
* @param ctx Kernal context.
@@ -448,18 +448,19 @@ public class GridJobProcessor extends GridProcessorAdapter {
else if (!nodeId.equals(taskNodeId))
err = "Received job siblings response from unexpected node [taskNodeId=" + taskNodeId +
", nodeId=" + nodeId + ']';
- else
+ else {
// Sender and message type are fine.
res = (GridJobSiblingsResponse)msg;
- if (res.jobSiblings() == null) {
- try {
- res.unmarshalSiblings(marsh);
- }
- catch (IgniteCheckedException e) {
- U.error(log, "Failed to unmarshal job siblings.", e);
+ if (res.jobSiblings() == null) {
+ try {
+ res.unmarshalSiblings(marsh);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to unmarshal job siblings.", e);
- err = e.getMessage();
+ err = e.getMessage();
+ }
}
}
@@ -830,7 +831,8 @@ public class GridJobProcessor extends GridProcessorAdapter {
if (w == null)
throw new NoSuchElementException();
- org.apache.ignite.spi.collision.CollisionJobContext ret = new CollisionJobContext(w, false);
+ org.apache.ignite.spi.collision.CollisionJobContext ret =
+ new CollisionJobContext(w, false);
w = null;
@@ -953,16 +955,14 @@ public class GridJobProcessor extends GridProcessorAdapter {
GridJobWorker job = null;
- rwLock.readLock();
-
- try {
- if (stopping) {
- if (log.isDebugEnabled())
- log.debug("Received job execution request while stopping this node (will ignore): " + req);
+ if (!rwLock.tryReadLock()) {
+ if (log.isDebugEnabled())
+ log.debug("Received job execution request while stopping this node (will ignore): " + req);
- return;
- }
+ return;
+ }
+ try {
long endTime = req.getCreateTime() + req.getTimeout();
// Account for overflow.
@@ -1172,7 +1172,7 @@ public class GridJobProcessor extends GridProcessorAdapter {
* @param ses Session.
*/
public void currentTaskSession(ComputeTaskSession ses) {
- currentSess.set(ses);
+ currSess.set(ses);
}
/**
@@ -1195,7 +1195,7 @@ public class GridJobProcessor extends GridProcessorAdapter {
if (!ctx.security().enabled())
return null;
- ComputeTaskSession ses = currentSess.get();
+ ComputeTaskSession ses = currSess.get();
if (ses == null)
return null;
@@ -1404,16 +1404,14 @@ public class GridJobProcessor extends GridProcessorAdapter {
*/
@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter", "RedundantCast"})
private void processTaskSessionRequest(UUID nodeId, GridTaskSessionRequest req) {
- rwLock.readLock();
-
- try {
- if (stopping) {
- if (log.isDebugEnabled())
- log.debug("Received job session request while stopping grid (will ignore): " + req);
+ if (!rwLock.tryReadLock()) {
+ if (log.isDebugEnabled())
+ log.debug("Received job session request while stopping grid (will ignore): " + req);
- return;
- }
+ return;
+ }
+ try {
GridTaskSessionImpl ses = ctx.session().getSession(req.getSessionId());
if (ses == null) {
@@ -1557,16 +1555,14 @@ public class GridJobProcessor extends GridProcessorAdapter {
if (log.isDebugEnabled())
log.debug("Received external collision event.");
- rwLock.readLock();
-
- try {
- if (stopping) {
- if (log.isDebugEnabled())
- log.debug("Received external collision notification while stopping grid (will ignore).");
+ if (!rwLock.tryReadLock()) {
+ if (log.isDebugEnabled())
+ log.debug("Received external collision notification while stopping grid (will ignore).");
- return;
- }
+ return;
+ }
+ try {
handleCollisions();
}
finally {
@@ -1653,16 +1649,14 @@ public class GridJobProcessor extends GridProcessorAdapter {
updateJobMetrics();
}
else {
- rwLock.readLock();
-
- try {
- if (stopping) {
- if (log.isDebugEnabled())
- log.debug("Skipping collision handling on job finish (node is stopping).");
+ if (!rwLock.tryReadLock()) {
+ if (log.isDebugEnabled())
+ log.debug("Skipping collision handling on job finish (node is stopping).");
- return;
- }
+ return;
+ }
+ try {
handleCollisions();
}
finally {
@@ -1851,16 +1845,14 @@ public class GridJobProcessor extends GridProcessorAdapter {
}
if (handleCollisions) {
- rwLock.readLock();
-
- try {
- if (stopping) {
- if (log.isDebugEnabled())
- log.debug("Skipped collision handling on discovery event (node is stopping): " + evt);
+ if (!rwLock.tryReadLock()) {
+ if (log.isDebugEnabled())
+ log.debug("Skipped collision handling on discovery event (node is stopping): " + evt);
- return;
- }
+ return;
+ }
+ try {
if (!jobAlwaysActivate)
handleCollisions();
else if (metricsUpdateFreq > -1L)
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/test/java/org/apache/ignite/internal/GridTaskFailoverAffinityRunTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridTaskFailoverAffinityRunTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskFailoverAffinityRunTest.java
index f1ae478..3b33b83 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridTaskFailoverAffinityRunTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridTaskFailoverAffinityRunTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
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;
@@ -54,6 +55,8 @@ public class GridTaskFailoverAffinityRunTest extends GridCommonAbstractTest {
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+ ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setSharedMemoryPort(-1);
+
boolean client = clientMode && gridName.equals(getTestGridName(0));
if (client) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/705ee1ee/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
index 8f90dbd..1e3382d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
@@ -1761,7 +1761,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
log.error("Failed to wait for update.");
for (Ignite ignite : G.allGrids())
- dumpCacheDebugInfo(ignite);
+ ((IgniteKernal)ignite).dumpDebugInfo();
U.dumpThreads(log);
@@ -1801,7 +1801,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
log.error("Failed to wait for update.");
for (Ignite ignite : G.allGrids())
- dumpCacheDebugInfo(ignite);
+ ((IgniteKernal)ignite).dumpDebugInfo();
U.dumpThreads(log);