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/05/02 15:08:20 UTC
[19/26] ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-5075-cacheStart' into ignite-5075
Merge remote-tracking branch 'remotes/origin/ignite-5075-cacheStart' into ignite-5075
# Conflicts:
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/GridClusterStateProcessor.java
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c132480e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c132480e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c132480e
Branch: refs/heads/ignite-5075
Commit: c132480ea069f1bef4f63779a341b4a7f2991bcd
Parents: 91452b1 bf2360d
Author: sboikov <sb...@gridgain.com>
Authored: Tue May 2 13:18:07 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue May 2 13:18:07 2017 +0300
----------------------------------------------------------------------
RELEASE_NOTES.txt | 36 +
examples/rest/http-rest-example.php | 4 +-
.../discovery/GridDiscoveryManager.java | 11 +-
.../affinity/AffinityTopologyVersion.java | 9 +
.../cache/CacheAffinitySharedManager.java | 210 +--
.../CacheClientReconnectDiscoveryData.java | 128 ++
.../internal/processors/cache/CacheData.java | 169 +++
.../cache/CacheJoinNodeDiscoveryData.java | 128 ++
.../cache/CacheNodeCommonDiscoveryData.java | 79 ++
.../processors/cache/ClusterCachesInfo.java | 730 ++++++++++
.../cache/DynamicCacheChangeBatch.java | 83 +-
.../cache/DynamicCacheChangeRequest.java | 95 +-
.../cache/DynamicCacheDescriptor.java | 26 +-
.../processors/cache/ExchangeActions.java | 332 +++++
.../processors/cache/GridCacheContext.java | 27 +-
.../processors/cache/GridCacheIoManager.java | 9 +-
.../GridCachePartitionExchangeManager.java | 58 +-
.../processors/cache/GridCacheProcessor.java | 1243 +++++-------------
.../dht/GridDhtPartitionTopologyImpl.java | 10 +-
.../GridDhtPartitionsExchangeFuture.java | 128 +-
.../cache/query/GridCacheQueryManager.java | 3 +-
.../cluster/GridClusterStateProcessor.java | 21 +-
.../ignite/spi/discovery/DiscoveryDataBag.java | 18 +-
.../processors/cache/IgniteCacheStartTest.java | 183 +++
.../cache/IgniteDynamicCacheStartSelfTest.java | 2 +-
.../CacheLateAffinityAssignmentTest.java | 4 +-
.../loadtests/hashmap/GridCacheTestContext.java | 3 +
.../testsuites/IgniteCacheTestSuite4.java | 2 +
modules/hibernate-core/pom.xml | 8 +
pom.xml | 2 +-
30 files changed, 2442 insertions(+), 1319 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c132480e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
index b221e6f,48bc6da..532e3ea
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
@@@ -356,11 -349,28 +351,28 @@@ public class CacheAffinitySharedManager
assert old == null : old;
}
}
+ }
- boolean clientOnly = true;
+ /**
+ * Called on exchange initiated for cache start/stop request.
+ *
+ * @param fut Exchange future.
+ * @param crd Coordinator flag.
+ * @param exchActions Cache change requests.
+ * @throws IgniteCheckedException If failed.
+ * @return {@code True} if client-only exchange is needed.
+ */
+ public boolean onCacheChangeRequest(final GridDhtPartitionsExchangeFuture fut,
+ boolean crd,
+ ExchangeActions exchActions)
+ throws IgniteCheckedException
+ {
+ assert exchActions != null && !exchActions.empty() : exchActions;
+
+ updateCachesInfo(exchActions);
// Affinity did not change for existing caches.
- forAllCaches(crd && lateAffAssign, new IgniteInClosureX<GridAffinityAssignmentCache>() {
+ forAllCacheGroups(crd && lateAffAssign, new IgniteInClosureX<GridAffinityAssignmentCache>() {
@Override public void applyx(GridAffinityAssignmentCache aff) throws IgniteCheckedException {
if (fut.stopping(aff.cacheId()))
return;
@@@ -377,79 -384,75 +386,75 @@@
Integer cacheId = CU.cacheId(req.cacheName());
- if (req.start()) {
- cctx.cache().prepareCacheStart(req, fut.topologyVersion());
+ boolean startCache;
- if (fut.isCacheAdded(cacheId, fut.topologyVersion())) {
- if (fut.discoCache().cacheAffinityNodes(req.cacheName()).isEmpty())
- U.quietAndWarn(log, "No server nodes found for cache client: " + req.cacheName());
- }
+ NearCacheConfiguration nearCfg = null;
+
+ if (cctx.localNodeId().equals(req.initiatingNodeId())) {
+ startCache = true;
+
+ nearCfg = req.nearCacheConfiguration();
+ }
+ else {
+ startCache = cctx.cacheContext(action.descriptor().cacheId()) == null &&
+ CU.affinityNode(cctx.localNode(), req.startCacheConfiguration().getNodeFilter());
+ }
- if (!crd || !lateAffAssign) {
- GridCacheContext cacheCtx = cctx.cacheContext(cacheId);
+ if (startCache)
+ cctx.cache().prepareCacheStart(req, nearCfg, action.descriptor(), fut.topologyVersion());
- if (cacheCtx != null && !cacheCtx.isLocal()) {
- boolean clientCacheStarted =
- req.clientStartOnly() && req.initiatingNodeId().equals(cctx.localNodeId());
+ if (fut.isCacheAdded(cacheId, fut.topologyVersion())) {
+ if (fut.discoCache().cacheAffinityNodes(req.cacheName()).isEmpty())
+ U.quietAndWarn(log, "No server nodes found for cache client: " + req.cacheName());
+ }
- if (clientCacheStarted)
- initAffinity(cacheCtx.affinity().affinityCache(), fut, lateAffAssign);
- else if (!req.clientStartOnly()) {
- assert fut.topologyVersion().equals(cacheCtx.startTopologyVersion());
+ if (!crd || !lateAffAssign) {
+ GridCacheContext cacheCtx = cctx.cacheContext(cacheId);
- GridAffinityAssignmentCache aff = cacheCtx.affinity().affinityCache();
+ if (cacheCtx != null && !cacheCtx.isLocal()) {
+ boolean clientCacheStarted =
+ req.clientStartOnly() && req.initiatingNodeId().equals(cctx.localNodeId());
- assert aff.lastVersion().equals(AffinityTopologyVersion.NONE) : aff.lastVersion();
+ if (clientCacheStarted)
+ initAffinity(cacheCtx.affinity().affinityCache(), fut, lateAffAssign);
+ else if (!req.clientStartOnly()) {
+ assert fut.topologyVersion().equals(cacheCtx.cacheStartTopologyVersion());
- List<List<ClusterNode>> assignment = aff.calculate(fut.topologyVersion(),
- fut.discoveryEvent(), fut.discoCache());
+ GridAffinityAssignmentCache aff = cacheCtx.affinity().affinityCache();
- aff.initialize(fut.topologyVersion(), assignment);
- }
+ assert aff.lastVersion().equals(AffinityTopologyVersion.NONE) : aff.lastVersion();
+
+ List<List<ClusterNode>> assignment = aff.calculate(fut.topologyVersion(),
+ fut.discoveryEvent(), fut.discoCache());
+
+ aff.initialize(fut.topologyVersion(), assignment);
}
}
- else
- initStartedCacheOnCoordinator(fut, req.cacheDescriptor());
}
- else if (req.stop() || req.close()) {
- cctx.cache().blockGateway(req);
+ else
- initStartedCacheOnCoordinator(fut, cacheId);
++ initStartedCacheOnCoordinator(fut, req.cacheDescriptor());
+ }
- if (crd) {
- boolean rmvCache = false;
+ for (DynamicCacheChangeRequest req : exchActions.closeRequests(cctx.localNodeId())) {
+ Integer cacheId = CU.cacheId(req.cacheName());
- if (req.close() && req.initiatingNodeId().equals(cctx.localNodeId())) {
- GridCacheContext cacheCtx = cctx.cacheContext(cacheId);
+ cctx.cache().blockGateway(req);
- rmvCache = cacheCtx != null && !cacheCtx.affinityNode();
- }
- else if (req.stop())
- rmvCache = true;
-
- if (rmvCache) {
- // TODO IGNITE-5075.
- // CacheHolder cache = caches.remove(cacheId);
- //
- // if (cache != null) {
- // if (!req.stop()) {
- // assert !cache.client();
- //
- // cache = CacheHolder2.create(cctx,
- // cctx.cache().cacheDescriptor(cacheId),
- // fut,
- // cache.affinity());
- //
- // caches.put(cacheId, cache);
- // }
- // else {
- // if (stoppedCaches == null)
- // stoppedCaches = new HashSet<>();
- //
- // stoppedCaches.add(cache.cacheId());
- //
- // cctx.io().removeHandler(cacheId, GridDhtAffinityAssignmentResponse.class);
- // }
- // }
- }
+ if (crd) {
+ GridCacheContext cacheCtx = cctx.cacheContext(cacheId);
+
+ // Client cache was stopped, need create 'client' CacheHolder.
+ if (cacheCtx != null && !cacheCtx.affinityNode()) {
+ CacheHolder cache = caches.remove(cacheId);
+
+ assert !cache.client() : cache;
+
+ cache = CacheHolder2.create(cctx,
+ cctx.cache().cacheDescriptor(cacheId),
+ fut,
+ cache.affinity());
+
+ caches.put(cacheId, cache);
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c132480e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c132480e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheDescriptor.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c132480e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index a62badf,6cb8907..5652189
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@@ -291,8 -291,9 +294,10 @@@ public class GridCacheContext<K, V> imp
GridKernalContext ctx,
GridCacheSharedContext sharedCtx,
CacheConfiguration cacheCfg,
+ CacheGroupInfrastructure grp,
CacheType cacheType,
+ AffinityTopologyVersion cacheStartTopVer,
+ AffinityTopologyVersion locStartTopVer,
boolean affNode,
boolean updatesAllowed,
MemoryPolicy memPlc,
@@@ -320,9 -321,9 +325,11 @@@
assert ctx != null;
assert sharedCtx != null;
assert cacheCfg != null;
+ assert cacheStartTopVer != null : cacheCfg.getName();
+ assert locStartTopVer != null : cacheCfg.getName();
+ assert grp != null;
+
assert evtMgr != null;
assert storeMgr != null;
assert evictMgr != null;
@@@ -338,8 -339,9 +345,10 @@@
this.ctx = ctx;
this.sharedCtx = sharedCtx;
this.cacheCfg = cacheCfg;
+ this.grp = grp;
this.cacheType = cacheType;
+ this.locStartTopVer = locStartTopVer;
+ this.cacheStartTopVer = cacheStartTopVer;
this.affNode = affNode;
this.updatesAllowed = updatesAllowed;
this.depEnabled = ctx.deploy().enabled() && !cacheObjects().isBinaryEnabled(cacheCfg);
http://git-wip-us.apache.org/repos/asf/ignite/blob/c132480e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 7ef8af5,b4e4b14..a007170
--- 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
@@@ -738,51 -664,25 +667,25 @@@ public class GridCacheProcessor extend
// Initialize defaults.
initialize(cfg, cacheObjCtx);
- String cacheName = cfg.getName();
+ boolean template = cfg.getName().endsWith("*");
- if (cacheDescriptor(cfg.getName()) != null) {
- throw new IgniteCheckedException("Duplicate cache name found (check configuration and " +
- "assign unique name to each cache): " + cacheName);
- }
+ if (!template) {
+ if (caches.containsKey(cfg.getName())) {
+ throw new IgniteCheckedException("Duplicate cache name found (check configuration and " +
+ "assign unique name to each cache): " + cfg.getName());
+ }
- CacheType cacheType;
+ CacheType cacheType;
- if (CU.isUtilityCache(cfg.getName()))
+ if (CU.isUtilityCache(cfg.getName()))
- cacheType = CacheType.UTILITY;
- else if (internalCaches.contains(cfg.getName()))
- cacheType = CacheType.INTERNAL;
- else
- cacheType = CacheType.USER;
+ cacheType = CacheType.UTILITY;
+ else if (internalCaches.contains(cfg.getName()))
+ cacheType = CacheType.INTERNAL;
+ else
+ cacheType = CacheType.USER;
- if (cacheType != CacheType.USER && cfg.getMemoryPolicyName() == null)
- cfg.setMemoryPolicyName(sharedCtx.database().systemMemoryPolicyName());
-
- boolean template = cfg.getName() != null && cfg.getName().endsWith("*");
-
- DynamicCacheDescriptor desc = new DynamicCacheDescriptor(ctx,
- cfg,
- cacheType,
- 0,
- template,
- IgniteUuid.randomUuid(),
- new QuerySchema(cfg.getQueryEntities()));
-
- desc.locallyConfigured(true);
- desc.staticallyConfigured(true);
- desc.receivedFrom(ctx.localNodeId());
-
- if (!template) {
- cacheDescriptor(cfg.getName(), desc);
-
- ctx.discovery().setCacheFilter(
- cfg.getName(),
- cfg.getNodeFilter(),
- cfg.getNearConfiguration() != null && cfg.getCacheMode() == PARTITIONED,
- cfg.getCacheMode());
-
- ctx.discovery().addClientNode(cfg.getName(),
- ctx.localNodeId(),
- cfg.getNearConfiguration() != null);
+ if (cacheType != CacheType.USER && cfg.getMemoryPolicyName() == null)
+ cfg.setMemoryPolicyName(sharedCtx.database().systemMemoryPolicyName());
if (!cacheType.userCache())
stopSeq.addLast(cfg.getName());
@@@ -831,6 -766,6 +769,25 @@@
internalCaches.add(CU.ATOMICS_CACHE_NAME);
}
++ @Nullable public CacheGroupInfrastructure cacheGroup(int grpId) {
++ return cacheGrps.get(grpId);
++ }
++
++ public Collection<CacheGroupInfrastructure> cacheGroups() {
++ return cacheGrps.values();
++ }
++
++ private int nextCacheGroupId(CacheConfiguration ccfg) {
++ if (ccfg.getGroupName() != null) {
++ for (DynamicCacheDescriptor cacheDesc : registeredCaches.values()) {
++ if (ccfg.getGroupName().equals(cacheDesc.cacheConfiguration().getGroupName()))
++ return cacheDesc.groupId();
++ }
++ }
++
++ return registeredCaches.size() + 1;
++ }
++
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override public void onKernalStart(boolean activeOnStart) throws IgniteCheckedException {
@@@ -1464,9 -1333,10 +1356,11 @@@
* @throws IgniteCheckedException If failed to create cache.
*/
private GridCacheContext createCache(CacheConfiguration<?, ?> cfg,
+ CacheGroupInfrastructure grp,
@Nullable CachePluginManager pluginMgr,
CacheType cacheType,
+ AffinityTopologyVersion cacheStartTopVer,
+ AffinityTopologyVersion locStartTopVer,
CacheObjectContext cacheObjCtx,
boolean updatesAllowed)
throws IgniteCheckedException {
@@@ -1540,8 -1410,9 +1434,10 @@@
ctx,
sharedCtx,
cfg,
+ grp,
cacheType,
+ cacheStartTopVer,
+ locStartTopVer,
affNode,
updatesAllowed,
memPlc,
@@@ -1672,8 -1543,9 +1568,10 @@@
ctx,
sharedCtx,
cfg,
+ grp,
cacheType,
+ cacheStartTopVer,
+ locStartTopVer,
affNode,
true,
memPlc,
@@@ -1913,23 -1798,9 +1824,23 @@@
}
}
- return started;
+ return started != null ? started : Collections.<DynamicCacheDescriptor>emptyList();
}
+ private CacheGroupInfrastructure startCacheGroup(CacheConfiguration cfg0, int grpId) throws IgniteCheckedException {
+ CacheConfiguration ccfg = new CacheConfiguration(cfg0);
+
+ CacheGroupInfrastructure grp = new CacheGroupInfrastructure(grpId, sharedCtx, ccfg);
+
+ grp.start();
+
+ CacheGroupInfrastructure old = cacheGrps.put(grpId, grp);
+
+ assert old == null;
+
+ return grp;
+ }
+
/**
* @param cfg Start configuration.
* @param nearCfg Near configuration.
@@@ -1945,70 -1815,39 +1855,60 @@@
CacheConfiguration cfg,
NearCacheConfiguration nearCfg,
CacheType cacheType,
+ int grpId,
- boolean clientStartOnly,
- UUID initiatingNodeId,
IgniteUuid deploymentId,
- AffinityTopologyVersion topVer,
+ AffinityTopologyVersion cacheStartTopVer,
+ AffinityTopologyVersion exchTopVer,
@Nullable QuerySchema schema
) throws IgniteCheckedException {
- CacheConfiguration ccfg = new CacheConfiguration(cfg);
+ assert !caches.containsKey(cfg.getName()) : cfg.getName();
- IgnitePredicate nodeFilter = ccfg.getNodeFilter();
++ String grpName = cfg.getGroupName();
+
- ClusterNode locNode = ctx.discovery().localNode();
-
- boolean affNodeStart = !clientStartOnly && CU.affinityNode(locNode, nodeFilter);
- boolean clientNodeStart = locNode.id().equals(initiatingNodeId);
-
- if (sharedCtx.cacheContext(CU.cacheId(cfg.getName())) != null)
- return;
++ CacheGroupInfrastructure grp = null;
+
- if (affNodeStart || clientNodeStart || CU.isSystemCache(cfg.getName())) {
- String grpName = cfg.getGroupName();
++ if (grpName != null) {
++ for (CacheGroupInfrastructure grp0 : cacheGrps.values()) {
++ if (grp0.sharedGroup() && grpName.equals(grp0.groupName())) {
++ grp = grp0;
+
- CacheGroupInfrastructure grp = null;
-
- if (grpName != null) {
- for (CacheGroupInfrastructure grp0 : cacheGrps.values()) {
- if (grp0.sharedGroup() && grpName.equals(grp0.groupName())) {
- grp = grp0;
-
- break;
- }
++ break;
+ }
-
- if (grp == null)
- grp = startCacheGroup(cfg, grpId);
+ }
- else
++
++ if (grp == null)
+ grp = startCacheGroup(cfg, grpId);
++ }
++ else
++ grp = startCacheGroup(cfg, grpId);
+
- if (clientNodeStart && !affNodeStart) {
- if (nearCfg != null)
- ccfg.setNearConfiguration(nearCfg);
- else
- ccfg.setNearConfiguration(null);
- }
+ CacheConfiguration ccfg = new CacheConfiguration(cfg);
- CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(ccfg);
+ if (nearCfg != null)
+ ccfg.setNearConfiguration(nearCfg);
- GridCacheContext cacheCtx = createCache(ccfg, grp, null, cacheType, cacheObjCtx, true);
+ CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(ccfg);
- cacheCtx.startTopologyVersion(topVer);
+ GridCacheContext cacheCtx = createCache(ccfg,
++ grp,
+ null,
+ cacheType,
+ cacheStartTopVer,
+ exchTopVer,
+ cacheObjCtx,
+ true);
- cacheCtx.dynamicDeploymentId(deploymentId);
+ cacheCtx.dynamicDeploymentId(deploymentId);
- GridCacheAdapter cache = cacheCtx.cache();
+ GridCacheAdapter cache = cacheCtx.cache();
- sharedCtx.addCacheContext(cacheCtx);
+ sharedCtx.addCacheContext(cacheCtx);
- caches.put(cacheCtx.name(), cache);
+ caches.put(cacheCtx.name(), cache);
- startCache(cache, schema != null ? schema : new QuerySchema());
+ startCache(cache, schema != null ? schema : new QuerySchema());
- onKernalStart(cache);
- }
+ onKernalStart(cache);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/c132480e/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
index 57db6c3,6149586..e301993
--- a/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
@@@ -77,8 -78,9 +78,10 @@@ public class GridCacheTestContext<K, V
null
),
defaultCacheConfiguration(),
+ null,
CacheType.USER,
+ AffinityTopologyVersion.ZERO,
+ AffinityTopologyVersion.ZERO,
true,
true,
null,