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/06/06 15:06:05 UTC
[16/22] ignite git commit: ignite-5075 Implement logical 'cache
groups' sharing the same physical caches
ignite-5075 Implement logical 'cache groups' sharing the same physical caches
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/db85d166
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/db85d166
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/db85d166
Branch: refs/heads/ignite-5293
Commit: db85d166b066e39777093db9d7f44589a147276e
Parents: eaf8e14
Author: sboikov <sb...@gridgain.com>
Authored: Tue Jun 6 12:33:33 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue Jun 6 12:33:48 2017 +0300
----------------------------------------------------------------------
.../configuration/CacheConfiguration.java | 21 +
.../managers/communication/GridIoManager.java | 10 +-
.../internal/managers/discovery/DiscoCache.java | 30 +-
.../discovery/GridDiscoveryManager.java | 165 +-
.../ignite/internal/pagemem/PageSupport.java | 2 +-
.../affinity/GridAffinityAssignmentCache.java | 38 +-
.../cache/CacheAffinitySharedManager.java | 623 +--
.../CacheClientReconnectDiscoveryData.java | 66 +-
.../internal/processors/cache/CacheData.java | 22 +-
.../processors/cache/CacheGroupContext.java | 943 ++++
.../processors/cache/CacheGroupData.java | 147 +
.../processors/cache/CacheGroupDescriptor.java | 210 +
.../cache/CacheJoinNodeDiscoveryData.java | 4 +-
.../processors/cache/CacheMetricsImpl.java | 10 +-
.../cache/CacheNodeCommonDiscoveryData.java | 19 +
.../cache/CacheOffheapEvictionManager.java | 11 +-
.../processors/cache/ClusterCachesInfo.java | 533 ++-
.../cache/ClusterCachesReconnectResult.java | 61 +
.../cache/DynamicCacheDescriptor.java | 26 +
.../processors/cache/ExchangeActions.java | 86 +-
.../processors/cache/GridCacheAdapter.java | 60 +-
.../cache/GridCacheAffinityManager.java | 62 +-
.../processors/cache/GridCacheAttributes.java | 21 +-
.../cache/GridCacheClearAllRunnable.java | 2 +-
.../cache/GridCacheConcurrentMap.java | 55 +-
.../cache/GridCacheConcurrentMapImpl.java | 171 +-
.../processors/cache/GridCacheContext.java | 166 +-
.../processors/cache/GridCacheEntryInfo.java | 37 +-
.../processors/cache/GridCacheEventManager.java | 36 -
.../cache/GridCacheGroupIdMessage.java | 110 +
.../processors/cache/GridCacheIdMessage.java | 117 +
.../processors/cache/GridCacheIoManager.java | 282 +-
.../cache/GridCacheLocalConcurrentMap.java | 56 +-
.../processors/cache/GridCacheMapEntry.java | 73 +-
.../processors/cache/GridCacheMessage.java | 97 +-
.../GridCachePartitionExchangeManager.java | 291 +-
.../processors/cache/GridCachePreloader.java | 23 +-
.../cache/GridCachePreloaderAdapter.java | 46 +-
.../processors/cache/GridCacheProcessor.java | 391 +-
.../cache/GridCacheSharedContext.java | 4 +-
.../processors/cache/GridCacheTtlManager.java | 22 +-
.../processors/cache/GridCacheUtils.java | 42 +-
.../GridChangeGlobalStateMessageResponse.java | 20 +-
.../processors/cache/GridNoStorageCacheMap.java | 30 +-
.../cache/IgniteCacheOffheapManager.java | 143 +-
.../cache/IgniteCacheOffheapManagerImpl.java | 1194 +++--
.../cache/affinity/GridCacheAffinityImpl.java | 9 +-
.../processors/cache/database/CacheDataRow.java | 5 -
.../cache/database/CacheDataRowAdapter.java | 65 +-
.../cache/database/CacheSearchRow.java | 5 +
.../processors/cache/database/RowStore.java | 19 +-
.../cache/database/tree/BPlusTree.java | 41 +-
.../cache/database/tree/io/PageIO.java | 24 +
.../distributed/GridCacheTtlUpdateRequest.java | 4 +-
.../distributed/GridDistributedBaseMessage.java | 4 +-
.../GridDistributedCacheAdapter.java | 19 +-
.../GridDistributedTxFinishResponse.java | 28 +-
.../GridDistributedTxRemoteAdapter.java | 2 +
.../dht/GridCachePartitionedConcurrentMap.java | 76 +-
.../dht/GridClientPartitionTopology.java | 25 +-
.../dht/GridDhtAffinityAssignmentRequest.java | 10 +-
.../dht/GridDhtAffinityAssignmentResponse.java | 12 +-
.../dht/GridDhtAssignmentFetchFuture.java | 20 +-
.../distributed/dht/GridDhtCacheAdapter.java | 363 +-
.../distributed/dht/GridDhtCacheEntry.java | 16 +-
.../cache/distributed/dht/GridDhtGetFuture.java | 8 +-
.../distributed/dht/GridDhtGetSingleFuture.java | 41 +-
.../distributed/dht/GridDhtLocalPartition.java | 413 +-
.../distributed/dht/GridDhtLockResponse.java | 2 +-
.../dht/GridDhtPartitionTopology.java | 9 +-
.../dht/GridDhtPartitionTopologyImpl.java | 284 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 37 +-
.../dht/GridDhtTxFinishResponse.java | 14 +-
.../dht/GridDhtTxOnePhaseCommitAckRequest.java | 16 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 4 +-
.../dht/GridPartitionedGetFuture.java | 2 +-
.../dht/GridPartitionedSingleGetFuture.java | 6 +-
.../GridDhtAtomicAbstractUpdateRequest.java | 4 +-
.../dht/atomic/GridDhtAtomicCache.java | 68 +-
.../dht/atomic/GridDhtAtomicCacheEntry.java | 53 -
.../GridDhtAtomicDeferredUpdateResponse.java | 4 +-
.../dht/atomic/GridDhtAtomicNearResponse.java | 4 +-
.../dht/atomic/GridDhtAtomicUpdateResponse.java | 5 +-
.../GridNearAtomicAbstractUpdateRequest.java | 4 +-
.../GridNearAtomicCheckUpdateRequest.java | 4 +-
.../atomic/GridNearAtomicUpdateResponse.java | 4 +-
.../dht/colocated/GridDhtColocatedCache.java | 24 +-
.../colocated/GridDhtColocatedCacheEntry.java | 52 -
.../dht/preloader/GridDhtForceKeysFuture.java | 15 +-
.../dht/preloader/GridDhtForceKeysRequest.java | 4 +-
.../dht/preloader/GridDhtForceKeysResponse.java | 8 +-
.../GridDhtPartitionDemandMessage.java | 12 +-
.../dht/preloader/GridDhtPartitionDemander.java | 207 +-
.../dht/preloader/GridDhtPartitionSupplier.java | 58 +-
.../GridDhtPartitionSupplyMessage.java | 29 +-
.../GridDhtPartitionsAbstractMessage.java | 26 +-
.../GridDhtPartitionsExchangeFuture.java | 230 +-
.../preloader/GridDhtPartitionsFullMessage.java | 60 +-
.../GridDhtPartitionsSingleMessage.java | 40 +-
.../GridDhtPartitionsSingleRequest.java | 7 +-
.../dht/preloader/GridDhtPreloader.java | 410 +-
.../distributed/near/GridNearAtomicCache.java | 2 +-
.../distributed/near/GridNearCacheAdapter.java | 16 +-
.../distributed/near/GridNearGetRequest.java | 4 +-
.../distributed/near/GridNearGetResponse.java | 4 +-
.../near/GridNearOptimisticTxPrepareFuture.java | 21 +-
.../near/GridNearSingleGetRequest.java | 4 +-
.../near/GridNearSingleGetResponse.java | 6 +-
.../near/GridNearTransactionalCache.java | 4 +-
.../near/GridNearTxFinishResponse.java | 14 +-
.../processors/cache/local/GridLocalCache.java | 21 +-
.../local/atomic/GridLocalAtomicCache.java | 4 +-
.../query/GridCacheDistributedQueryManager.java | 10 +-
.../cache/query/GridCacheQueryManager.java | 4 +-
.../cache/query/GridCacheQueryRequest.java | 3 +-
.../cache/query/GridCacheQueryResponse.java | 4 +-
.../CacheContinuousQueryBatchAck.java | 4 +-
.../CacheContinuousQueryEventBuffer.java | 5 +-
.../continuous/CacheContinuousQueryHandler.java | 66 +-
.../CacheContinuousQueryListener.java | 20 +
.../continuous/CacheContinuousQueryManager.java | 59 +-
.../query/continuous/CounterSkipContext.java | 78 +
.../cache/transactions/IgniteTxAdapter.java | 2 +-
.../cache/transactions/IgniteTxEntry.java | 4 +-
.../cache/transactions/IgniteTxHandler.java | 26 +-
.../transactions/IgniteTxLocalAdapter.java | 2 +
.../cache/transactions/TxLocksRequest.java | 20 +-
.../cache/transactions/TxLocksResponse.java | 28 +-
.../cluster/GridClusterStateProcessor.java | 6 +-
.../processors/query/GridQueryProcessor.java | 3 +-
.../internal/processors/query/QueryUtils.java | 29 +-
.../visor/cache/VisorCachePartitionsTask.java | 2 +-
.../CacheAtomicSingleMessageCountSelfTest.java | 2 +-
.../cache/CacheDeferredDeleteQueueTest.java | 2 +-
...cheDhtLocalPartitionAfterRemoveSelfTest.java | 2 +-
...CacheExchangeMessageDuplicatedStateTest.java | 54 +-
.../cache/CacheOffheapMapEntrySelfTest.java | 9 +-
.../GridCacheConditionalDeploymentSelfTest.java | 18 +
.../processors/cache/GridCacheLeakTest.java | 3 +-
.../GridCacheOrderedPreloadingSelfTest.java | 14 +-
.../cache/GridCacheTtlManagerSelfTest.java | 3 +-
.../processors/cache/IgniteCacheGroupsTest.java | 4301 ++++++++++++++++++
.../cache/IgniteCachePeekModesAbstractTest.java | 2 +-
.../processors/cache/IgniteCacheStartTest.java | 5 +-
.../cache/IgniteOnePhaseCommitInvokeTest.java | 4 +-
...ologyValidatorAbstractTxCacheGroupsTest.java | 126 +
...opologyValidatorCacheGroupsAbstractTest.java | 131 +
...niteTopologyValidatorGridSplitCacheTest.java | 8 +-
...torNearPartitionedAtomicCacheGroupsTest.java | 31 +
...lidatorNearPartitionedTxCacheGroupsTest.java | 31 +
...lidatorPartitionedAtomicCacheGroupsTest.java | 45 +
...gyValidatorPartitionedTxCacheGroupsTest.java | 31 +
...alidatorReplicatedAtomicCacheGroupsTest.java | 45 +
...ogyValidatorReplicatedTxCacheGroupsTest.java | 31 +
.../IgniteTxStoreExceptionAbstractSelfTest.java | 4 +-
.../GridCacheBinaryObjectsAbstractSelfTest.java | 2 +-
.../GridCacheQueueCleanupSelfTest.java | 13 +-
.../GridCacheSetAbstractSelfTest.java | 17 +-
.../GridCacheSetFailoverAbstractSelfTest.java | 6 +-
.../IgnitePartitionedQueueNoBackupsTest.java | 6 +-
.../IgnitePartitionedSetNoBackupsSelfTest.java | 6 +-
.../CacheDiscoveryDataConcurrentJoinTest.java | 17 +
.../CacheLateAffinityAssignmentTest.java | 6 +-
...eCacheGroupsPartitionLossPolicySelfTest.java | 355 ++
.../IgniteCachePartitionLossPolicySelfTest.java | 19 +-
.../IgniteCacheReadFromBackupTest.java | 5 +-
...sabledMultiNodeWithGroupFullApiSelfTest.java | 35 +
.../atomic/IgniteCacheAtomicProtocolTest.java | 7 +-
...AtomicMultiNodeWithGroupFullApiSelfTest.java | 34 +
...nabledMultiNodeWithGroupFullApiSelfTest.java | 35 +
.../near/GridCacheNearReadersSelfTest.java | 4 +-
...tionedMultiNodeWithGroupFullApiSelfTest.java | 34 +
.../GridCacheReplicatedPreloadSelfTest.java | 3 +-
.../IgniteCacheClientNearCacheExpiryTest.java | 20 +-
.../expiry/IgniteCacheTtlCleanupSelfTest.java | 2 +-
...IgniteCacheJdbcBlobStoreNodeRestartTest.java | 3 +
...acheLocalAtomicWithGroupFullApiSelfTest.java | 34 +
.../local/GridCacheLocalFullApiSelfTest.java | 1 -
.../GridCacheLocalWithGroupFullApiSelfTest.java | 34 +
...nuousQueryConcurrentPartitionUpdateTest.java | 229 +-
...ContinuousQueryFailoverAbstractSelfTest.java | 7 +-
.../TxOptimisticDeadlockDetectionTest.java | 2 +-
.../TxPessimisticDeadlockDetectionTest.java | 2 +-
.../loadtests/hashmap/GridCacheTestContext.java | 6 +-
.../communication/GridCacheMessageSelfTest.java | 30 +
.../testframework/junits/GridAbstractTest.java | 17 +
.../junits/common/GridCommonAbstractTest.java | 64 +
.../IgniteCacheFullApiSelfTestSuite.java | 13 +
.../testsuites/IgniteCacheTestSuite3.java | 3 +
.../testsuites/IgniteCacheTestSuite5.java | 2 +
.../IgniteTopologyValidatorTestSuite.java | 14 +
.../query/h2/database/H2PkHashIndex.java | 4 +-
.../query/h2/database/H2RowFactory.java | 2 +-
.../processors/query/h2/database/H2Tree.java | 6 +-
.../query/h2/database/H2TreeIndex.java | 7 +-
.../query/h2/ddl/DdlStatementsProcessor.java | 2 +-
.../h2/twostep/GridReduceQueryExecutor.java | 12 +-
.../cache/IgniteCacheGroupsSqlTest.java | 312 ++
.../cache/IgniteCacheNoClassQuerySelfTest.java | 20 -
.../IgniteCacheGroupsCompareQueryTest.java | 45 +
...teCacheGroupsSqlDistributedJoinSelfTest.java | 30 +
...roupsSqlSegmentedIndexMultiNodeSelfTest.java | 28 +
...iteCacheGroupsSqlSegmentedIndexSelfTest.java | 37 +
.../query/IgniteSqlDistributedJoinSelfTest.java | 2 +-
.../query/h2/sql/BaseH2CompareQueryTest.java | 35 +-
.../IgniteCacheQuerySelfTestSuite2.java | 9 +
.../IgniteCacheWithIndexingTestSuite.java | 3 +
.../yardstick/IgniteBenchmarkArguments.java | 27 +-
.../org/apache/ignite/yardstick/IgniteNode.java | 3 +
.../cache/IgniteCacheAbstractBenchmark.java | 158 +
.../yardstick/cache/IgniteGetAllBenchmark.java | 3 +
.../cache/IgniteGetAllPutAllTxBenchmark.java | 2 +
.../cache/IgniteGetAndPutBenchmark.java | 2 +
.../cache/IgniteGetAndPutTxBenchmark.java | 2 +
.../yardstick/cache/IgniteGetBenchmark.java | 35 +-
.../IgniteGetEntriesPutAllTxBenchmark.java | 2 +
.../yardstick/cache/IgniteInvokeBenchmark.java | 2 +
.../cache/IgniteInvokeTxBenchmark.java | 2 +
.../IgniteInvokeWithInjectionBenchmark.java | 2 +
.../yardstick/cache/IgnitePutAllBenchmark.java | 2 +
.../IgnitePutAllSerializableTxBenchmark.java | 2 +
.../yardstick/cache/IgnitePutBenchmark.java | 2 +
.../cache/IgnitePutGetBatchBenchmark.java | 2 +
.../yardstick/cache/IgnitePutGetBenchmark.java | 2 +
.../cache/IgnitePutGetEntryBenchmark.java | 2 +
.../cache/IgnitePutGetEntryTxBenchmark.java | 2 +
.../cache/IgnitePutGetTxBatchBenchmark.java | 2 +
.../cache/IgnitePutGetTxBenchmark.java | 2 +
...IgnitePutIfAbsentIndexedValue1Benchmark.java | 2 +
.../cache/IgnitePutIndexedValue1Benchmark.java | 2 +
.../cache/IgnitePutIndexedValue2Benchmark.java | 2 +
.../cache/IgnitePutIndexedValue8Benchmark.java | 2 +
.../IgnitePutRandomValueSizeBenchmark.java | 2 +
.../cache/IgnitePutRemoveBenchmark.java | 2 +
.../yardstick/cache/IgnitePutTxBenchmark.java | 2 +
.../cache/IgnitePutTxImplicitBenchmark.java | 2 +
.../cache/IgnitePutTxPrimaryOnlyBenchmark.java | 2 +
.../IgnitePutTxSkipLocalBackupBenchmark.java | 2 +
.../cache/IgnitePutValue8Benchmark.java | 2 +
.../IgniteReplaceIndexedValue1Benchmark.java | 2 +
.../cache/IgniteScanQueryBenchmark.java | 88 +
.../cache/IgniteSqlQueryBenchmark.java | 16 +-
.../IgniteSqlQueryDistributedJoinBenchmark.java | 24 +-
.../cache/IgniteSqlQueryJoinBenchmark.java | 14 +-
.../cache/IgniteSqlQueryPutBenchmark.java | 8 +-
.../IgniteSqlQueryPutSeparatedBenchmark.java | 4 +
.../cache/jdbc/JdbcPutGetBenchmark.java | 4 +-
.../IgniteCacheRandomOperationBenchmark.java | 42 +-
.../yardstick/cache/load/model/ModelUtil.java | 3 +
249 files changed, 13546 insertions(+), 3450 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/db85d166/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
index 21f2fba..67b7590 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
@@ -202,6 +202,9 @@ public class CacheConfiguration<K, V> extends MutableConfiguration<K, V> {
/** Cache name. */
private String name;
+ /** Cache group name. */
+ private String grpName;
+
/** Name of {@link MemoryPolicyConfiguration} for this cache */
private String memPlcName;
@@ -408,6 +411,7 @@ public class CacheConfiguration<K, V> extends MutableConfiguration<K, V> {
evictFilter = cc.getEvictionFilter();
evictPlc = cc.getEvictionPolicy();
expiryPolicyFactory = cc.getExpiryPolicyFactory();
+ grpName = cc.getGroupName();
indexedTypes = cc.getIndexedTypes();
interceptor = cc.getInterceptor();
invalidate = cc.isInvalidate();
@@ -455,6 +459,23 @@ public class CacheConfiguration<K, V> extends MutableConfiguration<K, V> {
}
/**
+ * @return Cache group name.
+ */
+ public String getGroupName() {
+ return grpName;
+ }
+
+ /**
+ * @param grpName Cache group name.
+ * @return {@code this} for chaining.
+ */
+ public CacheConfiguration<K, V> setGroupName(String grpName) {
+ this.grpName = grpName;
+
+ return this;
+ }
+
+ /**
* Cache name or {@code null} if not provided, then this will be considered a default
* cache which can be accessed via {@link Ignite#cache(String)} method. Otherwise, if name
* is provided, the cache will be accessed via {@link Ignite#cache(String)} method.
http://git-wip-us.apache.org/repos/asf/ignite/blob/db85d166/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
index 698baf8..1efc4aa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
@@ -1153,10 +1153,14 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
pools.poolForPolicy(plc).execute(c);
}
catch (RejectedExecutionException e) {
- U.error(log, "Failed to process regular message due to execution rejection. Will attempt to process " +
- "message in the listener thread instead.", e);
+ if (!ctx.isStopping()) {
+ U.error(log, "Failed to process regular message due to execution rejection. Will attempt to process " +
+ "message in the listener thread instead.", e);
- c.run();
+ c.run();
+ }
+ else if (log.isDebugEnabled())
+ log.debug("Failed to process regular message due to execution rejection: " + msg);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/db85d166/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
index 5247ac1..22c2d07 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
@@ -69,7 +69,7 @@ public class DiscoCache {
/** Affinity cache nodes by cache name. */
@GridToStringInclude
- private final Map<Integer, List<ClusterNode>> affCacheNodes;
+ private final Map<Integer, List<ClusterNode>> cacheGrpAffNodes;
/** Node map. */
private final Map<UUID, ClusterNode> nodeMap;
@@ -91,7 +91,7 @@ public class DiscoCache {
* @param allNodesWithCaches All nodes with at least one cache configured.
* @param rmtNodesWithCaches Remote nodes with at least one cache configured.
* @param allCacheNodes Cache nodes by cache name.
- * @param affCacheNodes Affinity cache nodes by cache name.
+ * @param cacheGrpAffNodes Affinity nodes by cache group ID.
* @param nodeMap Node map.
* @param nearEnabledCaches Caches where at least one node has near cache enabled.
* @param alives Alive nodes.
@@ -105,7 +105,7 @@ public class DiscoCache {
List<ClusterNode> allNodesWithCaches,
List<ClusterNode> rmtNodesWithCaches,
Map<Integer, List<ClusterNode>> allCacheNodes,
- Map<Integer, List<ClusterNode>> affCacheNodes,
+ Map<Integer, List<ClusterNode>> cacheGrpAffNodes,
Map<UUID, ClusterNode> nodeMap,
Set<Integer> nearEnabledCaches,
Set<UUID> alives) {
@@ -118,7 +118,7 @@ public class DiscoCache {
this.allNodesWithCaches = allNodesWithCaches;
this.rmtNodesWithCaches = rmtNodesWithCaches;
this.allCacheNodes = allCacheNodes;
- this.affCacheNodes = affCacheNodes;
+ this.cacheGrpAffNodes = cacheGrpAffNodes;
this.nodeMap = nodeMap;
this.nearEnabledCaches = nearEnabledCaches;
this.alives.addAll(alives);
@@ -235,25 +235,11 @@ public class DiscoCache {
}
/**
- * Gets all nodes that have cache with given ID and should participate in affinity calculation. With
- * partitioned cache nodes with near-only cache do not participate in affinity node calculation.
- *
- * @param cacheName Cache name.
- * @return Collection of nodes.
- */
- public List<ClusterNode> cacheAffinityNodes(@Nullable String cacheName) {
- return cacheAffinityNodes(CU.cacheId(cacheName));
- }
-
- /**
- * Gets all nodes that have cache with given ID and should participate in affinity calculation. With
- * partitioned cache nodes with near-only cache do not participate in affinity node calculation.
- *
- * @param cacheId Cache ID.
- * @return Collection of nodes.
+ * @param grpId Cache group ID.
+ * @return All nodes that participate in affinity calculation.
*/
- public List<ClusterNode> cacheAffinityNodes(int cacheId) {
- return emptyIfNull(affCacheNodes.get(cacheId));
+ public List<ClusterNode> cacheGroupAffinityNodes(int grpId) {
+ return emptyIfNull(cacheGrpAffNodes.get(grpId));
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/db85d166/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 7c702c2..8c2bd8c 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
@@ -69,7 +69,7 @@ import org.apache.ignite.internal.managers.GridManagerAdapter;
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.DynamicCacheDescriptor;
+import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.jobmetrics.GridJobMetrics;
import org.apache.ignite.internal.processors.security.SecurityContext;
@@ -251,12 +251,15 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
new ConcurrentHashMap8<>();
/** Local node initialization event listeners. */
- private final Collection<IgniteInClosure<ClusterNode>> localNodeInitLsnrs = new ArrayList<>();
+ private final Collection<IgniteInClosure<ClusterNode>> locNodeInitLsnrs = new ArrayList<>();
/** Map of dynamic cache filters. */
private Map<String, CachePredicate> registeredCaches = new HashMap<>();
/** */
+ private Map<Integer, CacheGroupAffinity> registeredCacheGrps = new HashMap<>();
+
+ /** */
private final GridSpinBusyLock busyLock = new GridSpinBusyLock();
/** Received custom messages history. */
@@ -321,24 +324,55 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
+ *
+ */
+ public void onLocalNodeJoin() {
+ registeredCacheGrps.clear();
+ registeredCaches.clear();
+ }
+
+ /**
+ * @param grpDesc Cache group descriptor.
+ * @param filter Node filter.
+ * @param cacheMode Cache mode.
+ */
+ public void addCacheGroup(CacheGroupDescriptor grpDesc, IgnitePredicate<ClusterNode> filter, CacheMode cacheMode) {
+ CacheGroupAffinity old = registeredCacheGrps.put(grpDesc.groupId(),
+ new CacheGroupAffinity(grpDesc.cacheOrGroupName(), filter, cacheMode));
+
+ assert old == null : old;
+ }
+
+ /**
+ * @param grpDesc Cache group descriptor.
+ */
+ public void removeCacheGroup(CacheGroupDescriptor grpDesc) {
+ CacheGroupAffinity rmvd = registeredCacheGrps.remove(grpDesc.groupId());
+
+ assert rmvd != null : grpDesc.cacheOrGroupName();
+ }
+
+ /**
* Adds dynamic cache filter.
*
+ * @param grpId Cache group ID.
* @param cacheName Cache name.
- * @param filter Cache filter.
* @param nearEnabled Near enabled flag.
- * @param cacheMode Cache mode.
*/
public void setCacheFilter(
+ int grpId,
String cacheName,
- IgnitePredicate<ClusterNode> filter,
- boolean nearEnabled,
- CacheMode cacheMode
+ boolean nearEnabled
) {
if (!registeredCaches.containsKey(cacheName)) {
- if (cacheMode == CacheMode.REPLICATED)
+ CacheGroupAffinity grp = registeredCacheGrps.get(grpId);
+
+ assert grp != null : "Failed to find cache group [grpId=" + grpId + ", cache=" + cacheName + ']';
+
+ if (grp.cacheMode == CacheMode.REPLICATED)
nearEnabled = false;
-
- registeredCaches.put(cacheName, new CachePredicate(filter, nearEnabled, cacheMode));
+
+ registeredCaches.put(cacheName, new CachePredicate(grp, nearEnabled));
}
}
@@ -401,7 +435,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
}
- return res;
+ return res == null ? Collections.<String, Map<UUID,Boolean>>emptyMap() : res;
}
/**
@@ -493,7 +527,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
/** {@inheritDoc} */
@Override public void onLocalNodeInitialized(ClusterNode locNode) {
- for (IgniteInClosure<ClusterNode> lsnr : localNodeInitLsnrs)
+ for (IgniteInClosure<ClusterNode> lsnr : locNodeInitLsnrs)
lsnr.apply(locNode);
}
@@ -639,6 +673,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
locJoin = new GridFutureAdapter<>();
registeredCaches.clear();
+ registeredCacheGrps.clear();
for (AffinityTopologyVersion histVer : discoCacheHist.keySet()) {
Object rmvd = discoCacheHist.remove(histVer);
@@ -801,7 +836,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
* @param lsnr Listener to add.
*/
public void addLocalNodeInitializedEventListener(IgniteInClosure<ClusterNode> lsnr) {
- localNodeInitLsnrs.add(lsnr);
+ locNodeInitLsnrs.add(lsnr);
}
/**
@@ -1742,27 +1777,14 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
- * Gets cache nodes for cache with given name that participate in affinity calculation.
- *
- * @param cacheName Cache name.
- * @param topVer Topology version.
- * @return Collection of cache affinity nodes.
- */
- public Collection<ClusterNode> cacheAffinityNodes(@Nullable String cacheName, AffinityTopologyVersion topVer) {
- int cacheId = CU.cacheId(cacheName);
-
- return resolveDiscoCache(cacheId, topVer).cacheAffinityNodes(cacheId);
- }
-
- /**
* Gets cache nodes for cache with given ID that participate in affinity calculation.
*
- * @param cacheId Cache ID.
+ * @param grpId Cache group ID.
* @param topVer Topology version.
* @return Collection of cache affinity nodes.
*/
- public Collection<ClusterNode> cacheAffinityNodes(int cacheId, AffinityTopologyVersion topVer) {
- return resolveDiscoCache(cacheId, topVer).cacheAffinityNodes(cacheId);
+ public Collection<ClusterNode> cacheGroupAffinityNodes(int grpId, AffinityTopologyVersion topVer) {
+ return resolveDiscoCache(grpId, topVer).cacheGroupAffinityNodes(grpId);
}
/**
@@ -1825,7 +1847,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
if (!CU.isSystemCache(cacheName) && !CU.isIgfsCache(ctx.config(), cacheName) &&
pred != null && pred.cacheNode(node))
- caches.put(cacheName, pred.cacheMode);
+ caches.put(cacheName, pred.aff.cacheMode);
}
return caches;
@@ -1845,21 +1867,21 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
/**
* Gets discovery cache for given topology version.
*
- * @param cacheId Cache ID (participates in exception message).
+ * @param grpId Cache group ID (participates in exception message).
* @param topVer Topology version.
* @return Discovery cache.
*/
- private DiscoCache resolveDiscoCache(int cacheId, AffinityTopologyVersion topVer) {
+ private DiscoCache resolveDiscoCache(int grpId, AffinityTopologyVersion topVer) {
Snapshot snap = topSnap.get();
DiscoCache cache = AffinityTopologyVersion.NONE.equals(topVer) || topVer.equals(snap.topVer) ?
snap.discoCache : discoCacheHist.get(topVer);
if (cache == null) {
- DynamicCacheDescriptor desc = ctx.cache().cacheDescriptor(cacheId);
+ CacheGroupDescriptor desc = ctx.cache().cacheGroupDescriptors().get(grpId);
throw new IgniteException("Failed to resolve nodes topology [" +
- "cacheName=" + (desc != null ? desc.cacheConfiguration().getName() : "N/A") +
+ "cacheGrp=" + (desc != null ? desc.cacheOrGroupName() : "N/A") +
", topVer=" + topVer +
", history=" + discoCacheHist.keySet() +
", snap=" + snap +
@@ -2073,7 +2095,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
" [rmtNodes=" + rmtNodes + ", loc=" + loc + ']';
Map<Integer, List<ClusterNode>> allCacheNodes = U.newHashMap(allNodes.size());
- Map<Integer, List<ClusterNode>> affCacheNodes = U.newHashMap(allNodes.size());
+ Map<Integer, List<ClusterNode>> cacheGrpAffNodes = U.newHashMap(allNodes.size());
Set<ClusterNode> allNodesWithCaches = new TreeSet<>(GridNodeOrderComparator.INSTANCE);
Set<ClusterNode> rmtNodesWithCaches = new TreeSet<>(GridNodeOrderComparator.INSTANCE);
@@ -2085,6 +2107,20 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
assert node.order() != 0 : "Invalid node order [locNode=" + loc + ", node=" + node + ']';
assert !node.isDaemon();
+ for (Map.Entry<Integer, CacheGroupAffinity> e : registeredCacheGrps.entrySet()) {
+ CacheGroupAffinity grpAff = e.getValue();
+ Integer grpId = e.getKey();
+
+ if (CU.affinityNode(node, grpAff.cacheFilter)) {
+ List<ClusterNode> nodes = cacheGrpAffNodes.get(grpId);
+
+ if (nodes == null)
+ cacheGrpAffNodes.put(grpId, nodes = new ArrayList<>());
+
+ nodes.add(node);
+ }
+ }
+
for (Map.Entry<String, CachePredicate> entry : registeredCaches.entrySet()) {
String cacheName = entry.getKey();
CachePredicate filter = entry.getValue();
@@ -2100,9 +2136,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
addToMap(allCacheNodes, cacheName, node);
- if (filter.dataNode(node))
- addToMap(affCacheNodes, cacheName, node);
-
if (filter.nearNode(node))
nearEnabledCaches.add(CU.cacheId(cacheName));
}
@@ -2119,7 +2152,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
U.sealList(allNodesWithCaches),
U.sealList(rmtNodesWithCaches),
Collections.unmodifiableMap(allCacheNodes),
- Collections.unmodifiableMap(affCacheNodes),
+ Collections.unmodifiableMap(cacheGrpAffNodes),
Collections.unmodifiableMap(nodeMap),
Collections.unmodifiableSet(nearEnabledCaches),
alives);
@@ -2732,32 +2765,60 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
+ *
+ */
+ private static class CacheGroupAffinity {
+ /** */
+ private final String name;
+
+ /** Nodes filter. */
+ private final IgnitePredicate<ClusterNode> cacheFilter;
+
+ /** Cache mode. */
+ private final CacheMode cacheMode;
+
+ /**
+ * @param name Name.
+ * @param cacheFilter Node filter.
+ * @param cacheMode Cache mode.
+ */
+ CacheGroupAffinity(
+ String name,
+ IgnitePredicate<ClusterNode> cacheFilter,
+ CacheMode cacheMode) {
+ this.name = name;
+ this.cacheFilter = cacheFilter;
+ this.cacheMode = cacheMode;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "CacheGroupAffinity [name=" + name + ']';
+ }
+ }
+
+ /**
* Cache predicate.
*/
private static class CachePredicate {
/** Cache filter. */
- private final IgnitePredicate<ClusterNode> cacheFilter;
+ private final CacheGroupAffinity aff;
/** If near cache is enabled on data nodes. */
private final boolean nearEnabled;
- /** Cache mode. */
- private final CacheMode cacheMode;
-
/** Collection of client near nodes. */
private final ConcurrentHashMap<UUID, Boolean> clientNodes;
/**
- * @param cacheFilter Cache filter.
+ * @param aff Cache group affinity.
* @param nearEnabled Near enabled flag.
- * @param cacheMode Cache mode.
*/
- private CachePredicate(IgnitePredicate<ClusterNode> cacheFilter, boolean nearEnabled, CacheMode cacheMode) {
- assert cacheFilter != null;
+ private CachePredicate(CacheGroupAffinity aff, boolean nearEnabled) {
+ assert aff != null;
- this.cacheFilter = cacheFilter;
+ this.aff = aff;
this.nearEnabled = nearEnabled;
- this.cacheMode = cacheMode;
clientNodes = new ConcurrentHashMap<>();
}
@@ -2792,7 +2853,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
* @return {@code True} if this node is a data node for given cache.
*/
public boolean dataNode(ClusterNode node) {
- return CU.affinityNode(node, cacheFilter);
+ return CU.affinityNode(node, aff.cacheFilter);
}
/**
@@ -2800,7 +2861,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
* @return {@code True} if cache is accessible on the given node.
*/
public boolean cacheNode(ClusterNode node) {
- return !node.isDaemon() && (CU.affinityNode(node, cacheFilter) || clientNodes.containsKey(node.id()));
+ return !node.isDaemon() && (CU.affinityNode(node, aff.cacheFilter) || clientNodes.containsKey(node.id()));
}
/**
@@ -2808,7 +2869,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
* @return {@code True} if near cache is present on the given nodes.
*/
public boolean nearNode(ClusterNode node) {
- if (CU.affinityNode(node, cacheFilter))
+ if (CU.affinityNode(node, aff.cacheFilter))
return nearEnabled;
Boolean near = clientNodes.get(node.id());
http://git-wip-us.apache.org/repos/asf/ignite/blob/db85d166/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageSupport.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageSupport.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageSupport.java
index 0f39058..f4b2d96 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageSupport.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageSupport.java
@@ -98,7 +98,7 @@ public interface PageSupport {
* @param pageId Page ID.
* @param page Page pointer.
* @param walPlc {@code True} if page should be recorded to WAL, {@code false} if the page must not
-* be recorded and {@code null} for the default behavior.
+ * be recorded and {@code null} for the default behavior.
* @param dirtyFlag Determines whether the page was modified since the last checkpoint.
*/
public void writeUnlock(int cacheId, long pageId, long page, Boolean walPlc,
http://git-wip-us.apache.org/repos/asf/ignite/blob/db85d166/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 4830f06..b478462 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
@@ -60,11 +60,11 @@ public class GridAffinityAssignmentCache {
/** Cleanup history size. */
private final int MAX_HIST_SIZE = getInteger(IGNITE_AFFINITY_HISTORY_SIZE, 500);
- /** Cache name. */
- private final String cacheName;
+ /** Group name if specified or cache name. */
+ private final String cacheOrGrpName;
- /** */
- private final int cacheId;
+ /** Group ID. */
+ private final int grpId;
/** Number of backups. */
private final int backups;
@@ -115,7 +115,8 @@ public class GridAffinityAssignmentCache {
* Constructs affinity cached calculations.
*
* @param ctx Kernal context.
- * @param cacheName Cache name.
+ * @param cacheOrGrpName Cache or cache group name.
+ * @param grpId Group ID.
* @param aff Affinity function.
* @param nodeFilter Node filter.
* @param backups Number of backups.
@@ -123,7 +124,8 @@ public class GridAffinityAssignmentCache {
*/
@SuppressWarnings("unchecked")
public GridAffinityAssignmentCache(GridKernalContext ctx,
- String cacheName,
+ String cacheOrGrpName,
+ int grpId,
AffinityFunction aff,
IgnitePredicate<ClusterNode> nodeFilter,
int backups,
@@ -132,16 +134,16 @@ public class GridAffinityAssignmentCache {
assert ctx != null;
assert aff != null;
assert nodeFilter != null;
+ assert grpId != 0;
this.ctx = ctx;
this.aff = aff;
this.nodeFilter = nodeFilter;
- this.cacheName = cacheName;
+ this.cacheOrGrpName = cacheOrGrpName;
+ this.grpId = grpId;
this.backups = backups;
this.locCache = locCache;
- cacheId = CU.cacheId(cacheName);
-
log = ctx.log(GridAffinityAssignmentCache.class);
partsCnt = aff.partitions();
@@ -161,17 +163,17 @@ public class GridAffinityAssignmentCache {
}
/**
- * @return Cache name.
+ * @return Group name if it is specified, otherwise cache name.
*/
- public String cacheName() {
- return cacheName;
+ public String cacheOrGroupName() {
+ return cacheOrGrpName;
}
/**
- * @return Cache ID.
+ * @return Cache group ID.
*/
- public int cacheId() {
- return cacheId;
+ public int groupId() {
+ return grpId;
}
/**
@@ -269,7 +271,7 @@ public class GridAffinityAssignmentCache {
List<ClusterNode> sorted;
if (!locCache) {
- sorted = new ArrayList<>(discoCache.cacheAffinityNodes(cacheId()));
+ sorted = new ArrayList<>(discoCache.cacheGroupAffinityNodes(groupId()));
Collections.sort(sorted, GridNodeOrderComparator.INSTANCE);
}
@@ -432,7 +434,7 @@ public class GridAffinityAssignmentCache {
*/
public void dumpDebugInfo() {
if (!readyFuts.isEmpty()) {
- U.warn(log, "Pending affinity ready futures [cache=" + cacheName + ", lastVer=" + lastVersion() + "]:");
+ U.warn(log, "Pending affinity ready futures [grp=" + cacheOrGrpName + ", lastVer=" + lastVersion() + "]:");
for (AffinityReadyFuture fut : readyFuts.values())
U.warn(log, ">>> " + fut);
@@ -461,7 +463,7 @@ public class GridAffinityAssignmentCache {
if (cache == null) {
throw new IllegalStateException("Getting affinity for topology version earlier than affinity is " +
"calculated [locNode=" + ctx.discovery().localNode() +
- ", cache=" + cacheName +
+ ", grp=" + cacheOrGrpName +
", topVer=" + topVer +
", head=" + head.get().topologyVersion() +
", history=" + affCache.keySet() +