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 2017/06/09 14:20:47 UTC
[09/42] ignite git commit: ignite-5196 Need use concurrent collection
for GridDiscoveryManager.registeredCaches
ignite-5196 Need use concurrent collection for GridDiscoveryManager.registeredCaches
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/56768a2b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/56768a2b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/56768a2b
Branch: refs/heads/ignite-5267-merge-ea
Commit: 56768a2b2f50892537c91fa46db6b887660c9932
Parents: 3ab4b9a
Author: Igor Seliverstov <gv...@gmail.com>
Authored: Wed Jun 7 11:10:47 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Jun 7 11:10:47 2017 +0300
----------------------------------------------------------------------
.../discovery/GridDiscoveryManager.java | 32 ++--
.../handlers/cache/GridCacheCommandHandler.java | 4 +-
.../top/GridTopologyCommandHandler.java | 10 +-
.../ignite/internal/ClusterGroupSelfTest.java | 100 ++++++++++++
.../top/CacheTopologyCommandHandlerTest.java | 152 +++++++++++++++++++
.../testsuites/IgniteRestHandlerTestSuite.java | 2 +
6 files changed, 279 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/56768a2b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 8c2bd8c..c91ff74 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -54,6 +54,7 @@ import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cluster.ClusterMetrics;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.ClusterMetricsSnapshot;
@@ -70,6 +71,7 @@ import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
+import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.jobmetrics.GridJobMetrics;
import org.apache.ignite.internal.processors.security.SecurityContext;
@@ -254,7 +256,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
private final Collection<IgniteInClosure<ClusterNode>> locNodeInitLsnrs = new ArrayList<>();
/** Map of dynamic cache filters. */
- private Map<String, CachePredicate> registeredCaches = new HashMap<>();
+ private ConcurrentMap<String, CachePredicate> registeredCaches = new ConcurrentHashMap<>();
/** */
private Map<Integer, CacheGroupAffinity> registeredCacheGrps = new HashMap<>();
@@ -353,7 +355,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
- * Adds dynamic cache filter.
+ * Called from discovery thread. Adds dynamic cache filter.
*
* @param grpId Cache group ID.
* @param cacheName Cache name.
@@ -377,7 +379,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
- * Removes dynamic cache filter.
+ * Called from discovery thread. Removes dynamic cache filter.
*
* @param cacheName Cache name.
*/
@@ -404,7 +406,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
- * Removes near node ID from cache filter.
+ * Called from discovery thread. Removes near node ID from cache filter.
*
* @param cacheName Cache name.
* @param clientNodeId Near node ID.
@@ -419,6 +421,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
+ * Called from discovery thread.
+ *
* @return Client nodes map.
*/
public Map<String, Map<UUID, Boolean>> clientNodesMap() {
@@ -439,6 +443,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
+ * Called from discovery thread.
+ *
* @param leftNodeId Left node ID.
*/
private void updateClientNodes(UUID leftNodeId) {
@@ -1837,17 +1843,17 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
* @param node Node to check.
* @return Public cache names accessible on the given node.
*/
- public Map<String, CacheMode> nodeCaches(ClusterNode node) {
- Map<String, CacheMode> caches = U.newHashMap(registeredCaches.size());
+ public Map<String, CacheConfiguration> nodePublicCaches(ClusterNode node) {
+ Map<String, CacheConfiguration> caches = U.newHashMap(registeredCaches.size());
- for (Map.Entry<String, CachePredicate> entry : registeredCaches.entrySet()) {
- String cacheName = entry.getKey();
+ for (DynamicCacheDescriptor cacheDesc : ctx.cache().cacheDescriptors().values()) {
+ if (!cacheDesc.cacheType().userCache())
+ continue;
- CachePredicate pred = entry.getValue();
+ CachePredicate p = registeredCaches.get(cacheDesc.cacheName());
- if (!CU.isSystemCache(cacheName) && !CU.isIgfsCache(ctx.config(), cacheName) &&
- pred != null && pred.cacheNode(node))
- caches.put(cacheName, pred.aff.cacheMode);
+ if (p != null && p.cacheNode(node))
+ caches.put(cacheDesc.cacheName(), cacheDesc.cacheConfiguration());
}
return caches;
@@ -2059,6 +2065,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
+ * Called from discovery thread.
+ *
* @param loc Local node.
* @param topSnapshot Topology snapshot.
* @return Newly created discovery cache.
http://git-wip-us.apache.org/repos/asf/ignite/blob/56768a2b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cache/GridCacheCommandHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cache/GridCacheCommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cache/GridCacheCommandHandler.java
index da260b1..bfc5282 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cache/GridCacheCommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cache/GridCacheCommandHandler.java
@@ -943,10 +943,10 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter {
boolean sameCaches = true;
- int hash = discovery.nodeCaches(F.first(subgrid)).hashCode();
+ Set<String> caches = discovery.nodePublicCaches(F.first(subgrid)).keySet();
for (int i = 1; i < subgrid.size(); i++) {
- if (hash != discovery.nodeCaches(subgrid.get(i)).hashCode()) {
+ if (!caches.equals(discovery.nodePublicCaches(subgrid.get(i)).keySet())) {
sameCaches = false;
break;
http://git-wip-us.apache.org/repos/asf/ignite/blob/56768a2b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/top/GridTopologyCommandHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/top/GridTopologyCommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/top/GridTopologyCommandHandler.java
index d9e023d..fa677c7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/top/GridTopologyCommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/top/GridTopologyCommandHandler.java
@@ -27,14 +27,12 @@ import java.util.Iterator;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cluster.ClusterMetrics;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.client.GridClientCacheMode;
-import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
import org.apache.ignite.internal.processors.port.GridPortRecord;
import org.apache.ignite.internal.processors.rest.GridRestCommand;
import org.apache.ignite.internal.processors.rest.GridRestProtocol;
@@ -217,14 +215,12 @@ public class GridTopologyCommandHandler extends GridRestCommandHandlerAdapter {
nodeBean.setTcpAddresses(nonEmptyList(node.<Collection<String>>attribute(ATTR_REST_TCP_ADDRS)));
nodeBean.setTcpHostNames(nonEmptyList(node.<Collection<String>>attribute(ATTR_REST_TCP_HOST_NAMES)));
- GridCacheProcessor cacheProc = ctx.cache();
-
- Map<String, CacheMode> nodeCaches = ctx.discovery().nodeCaches(node);
+ Map<String, CacheConfiguration> nodeCaches = ctx.discovery().nodePublicCaches(node);
Collection<GridClientCacheBean> caches = new ArrayList<>(nodeCaches.size());
- for (String cacheName : nodeCaches.keySet())
- caches.add(createCacheBean(cacheProc.cacheConfiguration(cacheName)));
+ for (CacheConfiguration ccfg : nodeCaches.values())
+ caches.add(createCacheBean(ccfg));
nodeBean.setCaches(caches);
http://git-wip-us.apache.org/repos/asf/ignite/blob/56768a2b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
index 142626b..9df561a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
@@ -19,14 +19,21 @@ package org.apache.ignite.internal;
import java.util.Collection;
import java.util.LinkedList;
+import java.util.Random;
+import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.Ignite;
import org.apache.ignite.Ignition;
import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonTest;
/**
@@ -274,6 +281,99 @@ public class ClusterGroupSelfTest extends ClusterGroupAbstractTest {
/**
* @throws Exception If failed.
*/
+ public void testForCacheNodesOnDynamicCacheCreateDestroy() throws Exception {
+ Random rnd = ThreadLocalRandom.current();
+
+ final AtomicReference<Exception> ex = new AtomicReference<>();
+
+ IgniteInternalFuture fut = runCacheCreateDestroyTask(ex);
+
+ while (!fut.isDone())
+ ignite.cluster().forCacheNodes("cache" + rnd.nextInt(16)).nodes();
+
+ if (ex.get() != null)
+ throw ex.get();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testForClientNodesOnDynamicCacheCreateDestroy() throws Exception {
+ Random rnd = ThreadLocalRandom.current();
+
+ final AtomicReference<Exception> ex = new AtomicReference<>();
+
+ IgniteInternalFuture fut = runCacheCreateDestroyTask(ex);
+
+ while (!fut.isDone())
+ ignite.cluster().forClientNodes("cache" + rnd.nextInt(16)).nodes();
+
+ if (ex.get() != null)
+ throw ex.get();
+ }
+
+ /**
+ * @param exHldr Exception holder.
+ * @return Task future.
+ */
+ private IgniteInternalFuture runCacheCreateDestroyTask(final AtomicReference<Exception> exHldr) {
+ final long deadline = System.currentTimeMillis() + 5000;
+
+ final AtomicInteger cntr = new AtomicInteger();
+
+ return GridTestUtils.runMultiThreadedAsync(new Runnable() {
+ @Override public void run() {
+ int startIdx = cntr.getAndAdd(4);
+ int idx = 0;
+ boolean start = true;
+
+ Set<String> caches = U.newHashSet(4);
+
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ if (start) {
+ caches.add("cache" + (startIdx + idx));
+ ignite.createCache("cache" + (startIdx + idx));
+ }
+ else {
+ ignite.destroyCache("cache" + (startIdx + idx));
+ caches.remove("cache" + (startIdx + idx));
+ }
+
+ if ((idx = (idx + 1) % 4) == 0)
+ start = !start;
+ }
+ catch (Exception e) {
+ addException(exHldr, e);
+
+ break;
+ }
+ }
+
+ for (String cache : caches) {
+ try {
+ ignite.destroyCache(cache);
+ }
+ catch (Exception e) {
+ addException(exHldr, e);
+ }
+ }
+ }
+ }, 4, "cache-start-destroy");
+ }
+
+ /**
+ * @param exHldr Exception holder.
+ * @param ex Exception.
+ */
+ private void addException(AtomicReference<Exception> exHldr, Exception ex) {
+ if (exHldr.get() != null || !exHldr.compareAndSet(null, ex))
+ exHldr.get().addSuppressed(ex);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testEmptyGroup() throws Exception {
ClusterGroup emptyGrp = ignite.cluster().forAttribute("nonExistent", "val");
http://git-wip-us.apache.org/repos/asf/ignite/blob/56768a2b/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/handlers/top/CacheTopologyCommandHandlerTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/handlers/top/CacheTopologyCommandHandlerTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/handlers/top/CacheTopologyCommandHandlerTest.java
new file mode 100644
index 0000000..5d7dfd7
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/handlers/top/CacheTopologyCommandHandlerTest.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.processors.rest.handlers.top;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.processors.rest.GridRestCommand;
+import org.apache.ignite.internal.processors.rest.request.GridRestTopologyRequest;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class CacheTopologyCommandHandlerTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration() throws Exception {
+ // Discovery config.
+ TcpDiscoverySpi disco = new TcpDiscoverySpi()
+ .setIpFinder(new TcpDiscoveryVmIpFinder(true))
+ .setJoinTimeout(5000);
+
+ // Cache config.
+ CacheConfiguration ccfg = new CacheConfiguration("cache*")
+ .setCacheMode(CacheMode.LOCAL)
+ .setAtomicityMode(CacheAtomicityMode.ATOMIC);
+
+ ConnectorConfiguration clnCfg = new ConnectorConfiguration()
+ .setHost("127.0.0.1");
+
+ return super.getConfiguration()
+ .setLocalHost("127.0.0.1")
+ .setConnectorConfiguration(clnCfg)
+ .setDiscoverySpi(disco)
+ .setCacheConfiguration(ccfg);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTopologyCommandOnDynamicCacheCreateDestroy() throws Exception {
+ GridRestTopologyRequest req = new GridRestTopologyRequest();
+ req.command(GridRestCommand.TOPOLOGY);
+
+ topologyCommandOnDynamicCacheCreateDestroy(startGrid(), req);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNodeCommandOnDynamicCacheCreateDestroy1() throws Exception {
+ Ignite node = startGrid();
+
+ GridRestTopologyRequest req = new GridRestTopologyRequest();
+ req.command(GridRestCommand.NODE);
+ req.nodeId(node.cluster().localNode().id());
+
+ topologyCommandOnDynamicCacheCreateDestroy(node, req);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNodeCommandOnDynamicCacheCreateDestroy2() throws Exception {
+ Ignite node = startGrid();
+
+ GridRestTopologyRequest req = new GridRestTopologyRequest();
+ req.command(GridRestCommand.NODE);
+ req.nodeIp("127.0.0.1");
+
+ topologyCommandOnDynamicCacheCreateDestroy(node, req);
+ }
+
+ /**
+ * @param node Ignite node.
+ * @param req Rest request.
+ * @throws Exception If failed.
+ */
+ private void topologyCommandOnDynamicCacheCreateDestroy(final Ignite node, GridRestTopologyRequest req) throws Exception {
+ GridTopologyCommandHandler hnd = new GridTopologyCommandHandler(((IgniteKernal)node).context());
+
+ final AtomicReference<Exception> ex = new AtomicReference<>();
+
+ final long deadline = System.currentTimeMillis() + 5000;
+
+ final AtomicInteger cntr = new AtomicInteger();
+
+ IgniteInternalFuture fut = GridTestUtils.runMultiThreadedAsync(new Runnable() {
+ @Override public void run() {
+ int startIdx = cntr.getAndAdd(4);
+ int idx = 0;
+ boolean start = true;
+
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ if (start)
+ node.createCache("cache" + (startIdx + idx));
+ else
+ node.destroyCache("cache" + (startIdx + idx));
+
+ if ((idx = (idx + 1) % 4) == 0)
+ start = !start;
+ }
+ catch (Exception e) {
+ if (ex.get() != null || !ex.compareAndSet(null, e))
+ ex.get().addSuppressed(e);
+
+ break;
+ }
+ }
+ }
+ }, 4, "cache-start-destroy");
+
+ while (!fut.isDone())
+ hnd.handleAsync(req).get();
+
+ if (ex.get() != null)
+ throw ex.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56768a2b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteRestHandlerTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteRestHandlerTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteRestHandlerTestSuite.java
index 6263e8b..f3e5828 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteRestHandlerTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteRestHandlerTestSuite.java
@@ -22,6 +22,7 @@ import org.apache.ignite.internal.processors.rest.handlers.cache.GridCacheAtomic
import org.apache.ignite.internal.processors.rest.handlers.cache.GridCacheCommandHandlerSelfTest;
import org.apache.ignite.internal.processors.rest.handlers.log.GridLogCommandHandlerTest;
import org.apache.ignite.internal.processors.rest.handlers.query.GridQueryCommandHandlerTest;
+import org.apache.ignite.internal.processors.rest.handlers.top.CacheTopologyCommandHandlerTest;
/**
* REST support tests.
@@ -38,6 +39,7 @@ public class IgniteRestHandlerTestSuite extends TestSuite {
suite.addTestSuite(GridCacheAtomicCommandHandlerSelfTest.class);
suite.addTestSuite(GridLogCommandHandlerTest.class);
suite.addTestSuite(GridQueryCommandHandlerTest.class);
+ suite.addTestSuite(CacheTopologyCommandHandlerTest.class);
return suite;
}