ignite-5578 Affinity for local join
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/62e3e70f Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/62e3e70f Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/62e3e70f Branch: refs/heads/ignite-5578 Commit: 62e3e70fc88db37d6b5c2b1f78c967fc4c9104db Parents: d47d9b7 Author: sboikov <[email protected]> Authored: Wed Jul 12 11:00:51 2017 +0300 Committer: sboikov <[email protected]> Committed: Wed Jul 12 11:12:05 2017 +0300 ---------------------------------------------------------------------- .../dht/GridClientPartitionTopology.java | 5 +++ .../dht/GridDhtPartitionTopology.java | 9 +++++ .../dht/GridDhtPartitionTopologyImpl.java | 40 ++++++++++++-------- .../GridCacheDatabaseSharedManager.java | 8 ++-- .../IgniteClusterActivateDeactivateTest.java | 2 +- 5 files changed, 43 insertions(+), 21 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/62e3e70f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java index f4ed517..c1a465d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java @@ -367,6 +367,11 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology { } /** {@inheritDoc} */ + @Override public GridDhtLocalPartition forceCreatePartition(int p) throws IgniteCheckedException { + throw new UnsupportedOperationException(); + } + + /** {@inheritDoc} */ @Override public GridDhtLocalPartition localPartition(int p) { return localPartition(p, AffinityTopologyVersion.NONE, false); } http://git-wip-us.apache.org/repos/asf/ignite/blob/62e3e70f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java index 5f76d12..caf3936 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java @@ -130,6 +130,15 @@ public interface GridDhtPartitionTopology { throws GridDhtInvalidPartitionException; /** + * Unconditionally creates partition during restore of persisted partition state. + * + * @param p Partition ID. + * @return Partition. + * @throws IgniteCheckedException If failed. + */ + public GridDhtLocalPartition forceCreatePartition(int p) throws IgniteCheckedException; + + /** * @param topVer Topology version at the time of creation. * @param p Partition ID. * @param create If {@code true}, then partition will be created if it's not there. http://git-wip-us.apache.org/repos/asf/ignite/blob/62e3e70f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java index 601da1b..9d16d90 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java @@ -135,9 +135,6 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology { /** */ private volatile AffinityTopologyVersion rebalancedTopVer = AffinityTopologyVersion.NONE; - /** */ - private volatile boolean treatAllPartAsLoc; - /** * @param ctx Cache shared context. * @param grp Cache group. @@ -421,14 +418,6 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology { /** {@inheritDoc} */ @Override public void beforeExchange(GridDhtPartitionsExchangeFuture exchFut, boolean affReady) throws IgniteCheckedException { - DiscoveryEvent discoEvt = exchFut.discoveryEvent(); - - treatAllPartAsLoc = exchFut.activateCluster() - || (discoEvt.type() == EventType.EVT_NODE_JOINED - && discoEvt.eventNode().isLocal() - && !ctx.kernalContext().clientNode() - ); - ClusterNode loc = ctx.localNode(); ctx.database().checkpointReadLock(); @@ -540,8 +529,6 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology { /** {@inheritDoc} */ @Override public boolean afterExchange(GridDhtPartitionsExchangeFuture exchFut) throws IgniteCheckedException { - treatAllPartAsLoc = false; - boolean changed = false; int num = grp.affinity().partitions(); @@ -692,6 +679,29 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology { return loc; } + /** {@inheritDoc} */ + @Override public GridDhtLocalPartition forceCreatePartition(int p) throws IgniteCheckedException { + lock.writeLock().lock(); + + try { + GridDhtLocalPartition part = locParts.get(p); + + if (part != null) + return part; + + part = new GridDhtLocalPartition(ctx, grp, p); + + locParts.set(p, part); + + ctx.pageStore().onPartitionCreated(grp.groupId(), p); + + return part; + } + finally { + lock.writeLock().unlock(); + } + } + /** * @param p Partition number. * @param topVer Topology version. @@ -731,7 +741,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology { if (loc != null && state == EVICTED) { locParts.set(p, loc = null); - if (!treatAllPartAsLoc && !belongs) + if (!belongs) throw new GridDhtInvalidPartitionException(p, "Adding entry to evicted partition " + "(often may be caused by inconsistent 'key.hashCode()' implementation) " + "[part=" + p + ", topVer=" + topVer + ", this.topVer=" + this.topVer + ']'); @@ -741,7 +751,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology { "[part=" + p + ", shouldBeMoving=" + loc.reload() + "]"); if (loc == null) { - if (!treatAllPartAsLoc && !belongs) + if (!belongs) throw new GridDhtInvalidPartitionException(p, "Creating partition which does not belong to " + "local node (often may be caused by inconsistent 'key.hashCode()' implementation) " + "[part=" + p + ", topVer=" + topVer + ", this.topVer=" + this.topVer + ']'); http://git-wip-us.apache.org/repos/asf/ignite/blob/62e3e70f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java index 1f56cc0..acff3c0 100755 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java @@ -96,7 +96,6 @@ import org.apache.ignite.internal.pagemem.wal.record.WALRecord; import org.apache.ignite.internal.pagemem.wal.record.delta.PageDeltaRecord; import org.apache.ignite.internal.pagemem.wal.record.delta.PartitionDestroyRecord; import org.apache.ignite.internal.pagemem.wal.record.delta.PartitionMetaStateRecord; -import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.CacheGroupContext; import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor; import org.apache.ignite.internal.processors.cache.GridCacheContext; @@ -1551,8 +1550,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan T2<Integer, Long> fromWal = partStates.get(new T2<>(grpId, i)); - GridDhtLocalPartition part = grp.topology() - .localPartition(i, AffinityTopologyVersion.NONE, true); + GridDhtLocalPartition part = grp.topology().forceCreatePartition(i); assert part != null; @@ -1610,10 +1608,10 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan /** * @param cacheCtx Cache context to apply an update. * @param dataEntry Data entry to apply. + * @throws IgniteCheckedException If failed to restore. */ private void applyUpdate(GridCacheContext cacheCtx, DataEntry dataEntry) throws IgniteCheckedException { - GridDhtLocalPartition locPart = cacheCtx.topology() - .localPartition(dataEntry.partitionId(), AffinityTopologyVersion.NONE, true); + GridDhtLocalPartition locPart = cacheCtx.topology().forceCreatePartition(dataEntry.partitionId()); switch (dataEntry.op()) { case CREATE: http://git-wip-us.apache.org/repos/asf/ignite/blob/62e3e70f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClusterActivateDeactivateTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClusterActivateDeactivateTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClusterActivateDeactivateTest.java index 8a604be..7bde012 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClusterActivateDeactivateTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClusterActivateDeactivateTest.java @@ -1082,7 +1082,7 @@ public class IgniteClusterActivateDeactivateTest extends GridCommonAbstractTest client = false; - // Start one more nodes while transition is in progress. + // Start more nodes while transition is in progress. IgniteInternalFuture startFut1 = GridTestUtils.runAsync(new Callable() { @Override public Object call() throws Exception { startGrid(8);
