You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/08/28 14:48:45 UTC
ignite git commit: IGNITE-9389 Avoid deadlock on cache#close(). -
Fixes #4626.
Repository: ignite
Updated Branches:
refs/heads/master 186b2d6f9 -> 13060f8e0
IGNITE-9389 Avoid deadlock on cache#close(). - Fixes #4626.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/13060f8e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/13060f8e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/13060f8e
Branch: refs/heads/master
Commit: 13060f8e0b6ce32321961c50eab079bc5afeda17
Parents: 186b2d6
Author: Denis Mekhanikov <dm...@gmail.com>
Authored: Tue Aug 28 17:46:29 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Tue Aug 28 17:46:29 2018 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheProcessor.java | 72 +++++++++-----------
.../cache/CacheStopAndDestroySelfTest.java | 68 ++++++++++++++++++
2 files changed, 100 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/13060f8e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 272aad4..eb64d9c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -2182,50 +2182,31 @@ public class GridCacheProcessor extends GridProcessorAdapter {
Set<Integer> closeCaches(Set<String> cachesToClose, boolean retClientCaches) {
Set<Integer> ids = null;
- boolean locked = false;
+ for (String cacheName : cachesToClose) {
+ blockGateway(cacheName, false, false);
- try {
- for (String cacheName : cachesToClose) {
- blockGateway(cacheName, false, false);
-
- GridCacheContext ctx = sharedCtx.cacheContext(CU.cacheId(cacheName));
-
- if (ctx == null)
- continue;
-
- if (retClientCaches && !ctx.affinityNode()) {
- if (ids == null)
- ids = U.newHashSet(cachesToClose.size());
-
- ids.add(ctx.cacheId());
- }
+ GridCacheContext ctx = sharedCtx.cacheContext(CU.cacheId(cacheName));
- if (!ctx.affinityNode() && !locked) {
- // Do not close client cache while requests processing is in progress.
- sharedCtx.io().writeLock();
-
- locked = true;
- }
+ if (ctx == null)
+ continue;
- if (!ctx.affinityNode() && ctx.transactional())
- sharedCtx.tm().rollbackTransactionsForCache(ctx.cacheId());
+ if (retClientCaches && !ctx.affinityNode()) {
+ if (ids == null)
+ ids = U.newHashSet(cachesToClose.size());
- closeCache(ctx, false);
+ ids.add(ctx.cacheId());
}
- return ids;
- }
- finally {
- if (locked)
- sharedCtx.io().writeUnlock();
+ closeCache(ctx);
}
+
+ return ids;
}
/**
* @param cctx Cache context.
- * @param destroy Destroy flag.
*/
- private void closeCache(GridCacheContext cctx, boolean destroy) {
+ private void closeCache(GridCacheContext cctx) {
if (cctx.affinityNode()) {
GridCacheAdapter<?, ?> cache = caches.get(cctx.name());
@@ -2234,21 +2215,32 @@ public class GridCacheProcessor extends GridProcessorAdapter {
jCacheProxies.put(cctx.name(), new IgniteCacheProxyImpl(cache.context(), cache, false));
}
else {
- jCacheProxies.remove(cctx.name());
-
cctx.gate().onStopped();
- sharedCtx.database().checkpointReadLock();
+ // Do not close client cache while requests processing is in progress.
+ sharedCtx.io().writeLock();
try {
- prepareCacheStop(cctx.name(), destroy);
+ if (!cctx.affinityNode() && cctx.transactional())
+ sharedCtx.tm().rollbackTransactionsForCache(cctx.cacheId());
+
+ jCacheProxies.remove(cctx.name());
+
+ sharedCtx.database().checkpointReadLock();
+
+ try {
+ prepareCacheStop(cctx.name(), false);
+ }
+ finally {
+ sharedCtx.database().checkpointReadUnlock();
+ }
+
+ if (!cctx.group().hasCaches())
+ stopCacheGroup(cctx.group().groupId());
}
finally {
- sharedCtx.database().checkpointReadUnlock();
+ sharedCtx.io().writeUnlock();
}
-
- if (!cctx.group().hasCaches())
- stopCacheGroup(cctx.group().groupId());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/13060f8e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
index 5eb8292..6239b52 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
@@ -18,11 +18,15 @@
package org.apache.ignite.internal.processors.cache;
import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import javax.cache.Cache;
import javax.cache.CacheManager;
import javax.cache.Caching;
import javax.cache.configuration.MutableConfiguration;
+import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
@@ -782,6 +786,70 @@ public class CacheStopAndDestroySelfTest extends GridCommonAbstractTest {
}
/**
+ * @throws Exception If failed.
+ */
+ public void testConcurrentUseAndCloseFromClient() throws Exception {
+ testConcurrentUseAndClose(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testConcurrentUseAndCloseFromServer() throws Exception {
+ testConcurrentUseAndClose(false);
+ }
+
+ /**
+ * @param isClient Should client or server be used during the test.
+ * @throws Exception If failed.
+ */
+ private void testConcurrentUseAndClose(boolean isClient) throws Exception {
+ int threads = 8;
+ int keys = 1000;
+ int iterations = 20;
+
+ startGrid(0);
+
+ IgniteConfiguration igniteCfg = getConfiguration(getTestIgniteInstanceName(1));
+ igniteCfg.setClientMode(isClient);
+ Ignite ignite = startGrid(optimize(igniteCfg));
+
+ ExecutorService execSrvc = Executors.newFixedThreadPool(threads);
+
+ for (int i = 0; i < threads; i++) {
+ execSrvc.execute(() -> {
+ while (!Thread.interrupted()) {
+ try {
+ IgniteCache<Integer, String> cache = ignite.getOrCreateCache("cache");
+
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ int key = random.nextInt(keys);
+
+ if (random.nextBoolean())
+ cache.put(key, Integer.toString(key));
+ else
+ cache.get(key);
+ }
+ catch (Exception ignore) {
+ }
+ }
+ });
+ }
+
+ for (int i = 0; i < iterations; i++) {
+ System.out.println("Iteration #" + (i + 1));
+
+ IgniteCache<Integer, String> cache = ignite.getOrCreateCache("cache");
+
+ cache.close();
+
+ Thread.sleep(100);
+ }
+
+ execSrvc.shutdownNow();
+ }
+
+ /**
* @param cache Cache.
* @throws Exception If failed.
*/