Merge remote-tracking branch 'origin/ignite-gg-8.0.1.ea7' into ignite…

…-gg-8.0.2.ea1 # Conflicts: # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java # modules/core/src/main/resources/META-INF/classnames.properties # modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
apache · Dec 12, 2016 · 47e0116 · 47e0116
2 parents eb6fdbf + 962daa8
commit 47e0116
Show file tree

Hide file tree

Showing 40 changed files with 559 additions and 257 deletions.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
@@ -46,6 +46,7 @@
 import org.apache.ignite.internal.processors.platform.PlatformEventFilterListener;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.P2;
+import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.T3;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiPredicate;
@@ -136,8 +137,8 @@ public GridEventConsumeHandler() {
     }
 
     /** {@inheritDoc} */
-    @Override public void updateCounters(AffinityTopologyVersion topVer, Map<UUID, Map<Integer, Long>> cntrsPerNode,
-        Map<Integer, Long> cntrs) {
+    @Override public void updateCounters(AffinityTopologyVersion topVer, Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode,
+        Map<Integer, T2<Long, Long>> cntrs) {
         // No-op.
     }
 

diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java
@@ -32,6 +32,7 @@
 import org.apache.ignite.internal.processors.continuous.GridContinuousBatchAdapter;
 import org.apache.ignite.internal.processors.continuous.GridContinuousHandler;
 import org.apache.ignite.internal.util.lang.GridPeerDeployAware;
+import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiPredicate;
@@ -125,8 +126,8 @@ public GridMessageListenHandler(GridMessageListenHandler orig) {
     }
 
     /** {@inheritDoc} */
-    @Override public void updateCounters(AffinityTopologyVersion topVer, Map<UUID, Map<Integer, Long>> cntrsPerNode,
-        Map<Integer, Long> cntrs) {
+    @Override public void updateCounters(AffinityTopologyVersion topVer, Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode,
+        Map<Integer, T2<Long, Long>> cntrs) {
         // No-op.
     }
 

diff --git a/...les/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java b/...les/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java
@@ -593,7 +593,7 @@ private long absolute(long relativePtr) {
 
             pageIdx &= idxMask;
 
-            long offset = pageIdx * sysPageSize;
+            long offset = ((long)pageIdx) * sysPageSize;
 
             return pagesBase + offset;
         }

diff --git a/...s/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java b/...s/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java
@@ -75,7 +75,7 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager {
      * @param partId Partition ID.
      * @throws IgniteCheckedException If failed to handle partition destroy callback.
      */
-    public void onPartitionDestroyed(int cacheId, int partId) throws IgniteCheckedException;
+    public void onPartitionDestroyed(int cacheId, int partId, int tag) throws IgniteCheckedException;
 
     /**
      * Reads a page for the given cache ID. Cache ID may be {@code 0} if the page is a meta page.
@@ -115,7 +115,7 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager {
      * @param pageBuf Page buffer to write.
      * @throws IgniteCheckedException If failed to write page.
      */
-    public void write(int cacheId, long pageId, ByteBuffer pageBuf) throws IgniteCheckedException;
+    public void write(int cacheId, long pageId, ByteBuffer pageBuf, int tag) throws IgniteCheckedException;
 
     /**
      * Gets page offset within the page store file.

diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/PageStore.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/PageStore.java
@@ -72,7 +72,7 @@ public interface PageStore {
      * @param pageBuf Page buffer to write.
      * @throws IgniteCheckedException If page writing failed (IO error occurred).
      */
-    public void write(long pageId, ByteBuffer pageBuf) throws IgniteCheckedException;
+    public void write(long pageId, ByteBuffer pageBuf, int tag) throws IgniteCheckedException;
 
     /**
      * Gets page offset within the store file.

diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheState.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheState.java
@@ -26,6 +26,9 @@ public enum CacheState {
      */
     INACTIVE,
 
+    /** Cache is inactive. But process of it activation in progress. */
+    ACTIVATING,
+
     /**
      * Cache is active and operations. There are no lost partitions.
      */

diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -86,6 +86,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.IgniteExternalizableExpiryPolicy;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
 import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxAdapter;
@@ -4555,9 +4556,7 @@ protected Object readResolve() throws ObjectStreamException {
 
     /** {@inheritDoc} */
     @Override public IgniteInternalFuture<?> rebalance() {
-        ctx.preloader().forcePreload();
-
-        return ctx.preloader().syncFuture();
+        return ctx.preloader().forceRebalance();
     }
 
     /** {@inheritDoc} */

diff --git 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
@@ -81,6 +81,7 @@
 import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
 import org.apache.ignite.internal.processors.closure.GridClosureProcessor;
 import org.apache.ignite.internal.processors.plugin.CachePluginManager;
+import org.apache.ignite.internal.processors.query.GridQueryProcessor;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
 import org.apache.ignite.internal.util.F0;
 import org.apache.ignite.internal.util.lang.GridFunc;
@@ -1952,6 +1953,13 @@ public boolean allowFastLocalRead(int part, List<ClusterNode> affNodes, Affinity
         return affinityNode() && rebalanceEnabled() && hasPartition(part, affNodes, topVer);
     }
 
+    /**
+     * @return {@code True} if fast eviction is allowed.
+     */
+    public boolean allowFastEviction() {
+        return shared().database().persistenceEnabled() && !GridQueryProcessor.isEnabled(cacheCfg);
+    }
+
     /**
      * @param part Partition.
      * @param affNodes Affinity nodes.

diff --git a/...n/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/...n/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -733,9 +733,13 @@ public void forceDummyExchange(boolean reassign,
      *
      * @param exchFut Exchange future.
      */
-    public void forcePreloadExchange(GridDhtPartitionsExchangeFuture exchFut) {
+    public IgniteInternalFuture<Boolean> forceRebalance(GridDhtPartitionsExchangeFuture exchFut) {
+        GridFutureAdapter<Boolean> fut = new GridFutureAdapter<>();
+
         exchWorker.addFuture(
-            new GridDhtPartitionsExchangeFuture(cctx, exchFut.discoveryEvent(), exchFut.exchangeId()));
+            new GridDhtPartitionsExchangeFuture(cctx, exchFut.discoveryEvent(), exchFut.exchangeId(), fut));
+
+        return fut;
     }
 
     /**
@@ -1794,7 +1798,8 @@ void addFuture(GridDhtPartitionsExchangeFuture exchFut) {
                                 Callable<Boolean> r = cacheCtx.preloader().addAssignments(assignsMap.get(cacheId),
                                     forcePreload,
                                     waitList,
-                                    cnt);
+                                    cnt,
+                                    exchFut.forcedRebalanceFuture());
 
                                 if (r != null) {
                                     U.log(log, "Cache rebalancing scheduled: [cache=" + cacheCtx.name() +

diff --git a/...es/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java b/...es/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
@@ -29,6 +29,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessageV2;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloaderAssignments;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
 import org.apache.ignite.lang.IgnitePredicate;
 import org.jetbrains.annotations.Nullable;
 
@@ -91,7 +92,8 @@ public interface GridCachePreloader {
     public Callable<Boolean> addAssignments(GridDhtPreloaderAssignments assignments,
         boolean forcePreload,
         Collection<String> caches,
-        int cnt);
+        int cnt,
+        @Nullable GridFutureAdapter<Boolean> forcedRebFut);
 
     /**
      * @param p Preload predicate.
@@ -151,9 +153,9 @@ public IgniteInternalFuture<Object> request(GridNearAtomicAbstractUpdateRequest
         AffinityTopologyVersion topVer);
 
     /**
-     * Force preload process.
+     * Force Rebalance process.
      */
-    public void forcePreload();
+    public IgniteInternalFuture<Boolean> forceRebalance();
 
     /**
      * Unwinds undeploys.

diff --git a/.../src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloaderAdapter.java b/.../src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloaderAdapter.java
@@ -32,6 +32,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloaderAssignments;
 import org.apache.ignite.internal.util.future.GridFinishedFuture;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
 import org.apache.ignite.lang.IgnitePredicate;
 import org.jetbrains.annotations.Nullable;
 
@@ -89,8 +90,8 @@ public GridCachePreloaderAdapter(GridCacheContext<?, ?> cctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void forcePreload() {
-        // No-op.
+    @Override public IgniteInternalFuture<Boolean> forceRebalance() {
+        return new GridFinishedFuture<>(true);
     }
 
     /** {@inheritDoc} */
@@ -167,7 +168,7 @@ public GridCachePreloaderAdapter(GridCacheContext<?, ?> cctx) {
 
     /** {@inheritDoc} */
     @Override public Callable<Boolean> addAssignments(GridDhtPreloaderAssignments assignments, boolean forcePreload,
-        Collection<String> caches, int cnt) {
+        Collection<String> caches, int cnt, @Nullable GridFutureAdapter<Boolean> forcedRebFut) {
         return null;
     }
 

diff --git a/.../main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java b/.../main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
@@ -23,11 +23,13 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import javax.cache.Cache;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.NodeStoppingException;
 import org.apache.ignite.internal.pagemem.FullPageId;
 import org.apache.ignite.internal.pagemem.PageIdUtils;
 import org.apache.ignite.internal.pagemem.PageMemory;
@@ -53,6 +55,7 @@
 import org.apache.ignite.internal.util.GridAtomicLong;
 import org.apache.ignite.internal.util.GridCloseableIteratorAdapter;
 import org.apache.ignite.internal.util.GridEmptyCloseableIterator;
+import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.lang.GridCloseableIterator;
 import org.apache.ignite.internal.util.lang.GridCursor;
 import org.apache.ignite.internal.util.lang.GridIterator;
@@ -93,6 +96,9 @@ public class IgniteCacheOffheapManagerImpl extends GridCacheManagerAdapter imple
     /** */
     private final GridAtomicLong globalRmvId = new GridAtomicLong(U.currentTimeMillis() * 1000_000);
 
+    /** */
+    private final GridSpinBusyLock busyLock = new GridSpinBusyLock();
+
     /** {@inheritDoc} */
     @Override public GridAtomicLong globalRemoveId() {
         return globalRmvId;
@@ -148,6 +154,13 @@ protected void initDataStructures() throws IgniteCheckedException {
             destroyCacheDataStructures(destroy);
     }
 
+    /** {@inheritDoc} */
+    @Override protected void onKernalStop0(boolean cancel) {
+        super.onKernalStop0(cancel);
+
+        busyLock.block();
+    }
+
     /**
      *
      */
@@ -845,69 +858,88 @@ public CacheDataStoreImpl(
             key.valueBytes(cctx.cacheObjectContext());
             val.valueBytes(cctx.cacheObjectContext());
 
-            rowStore.addRow(dataRow);
+            if (!busyLock.enterBusy())
+                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");
 
-            assert dataRow.link() != 0 : dataRow;
+            try {
+                rowStore.addRow(dataRow);
 
-            DataRow old = dataTree.put(dataRow);
+                assert dataRow.link() != 0 : dataRow;
 
-            if (old == null)
-                storageSize.incrementAndGet();
+                DataRow old = dataTree.put(dataRow);
 
-            if (indexingEnabled) {
-                GridCacheQueryManager qryMgr = cctx.queries();
+                if (old == null)
+                    storageSize.incrementAndGet();
 
-                assert qryMgr.enabled();
+                if (indexingEnabled) {
+                    GridCacheQueryManager qryMgr = cctx.queries();
 
-                if (old != null)
-                    qryMgr.store(key, p, old.value(), old.version(), val, ver, expireTime, dataRow.link());
-                else
-                    qryMgr.store(key, p, null, null, val, ver, expireTime, dataRow.link());
-            }
+                    assert qryMgr.enabled();
 
-            if (old != null) {
-                assert old.link() != 0 : old;
+                    if (old != null)
+                        qryMgr.store(key, p, old.value(), old.version(), val, ver, expireTime, dataRow.link());
+                    else
+                        qryMgr.store(key, p, null, null, val, ver, expireTime, dataRow.link());
+                }
 
-                if (pendingEntries != null && old.expireTime() != 0)
-                    pendingEntries.remove(new PendingRow(old.expireTime(), old.link()));
+                if (old != null) {
+                    assert old.link() != 0 : old;
 
-                rowStore.removeRow(old.link());
-            }
+                    if (pendingEntries != null && old.expireTime() != 0)
+                        pendingEntries.remove(new PendingRow(old.expireTime(), old.link()));
+
+                    rowStore.removeRow(old.link());
+                }
+
+                if (pendingEntries != null && expireTime != 0) {
+                    pendingEntries.put(new PendingRow(expireTime, dataRow.link()));
 
-            if (pendingEntries != null && expireTime != 0)
-                pendingEntries.put(new PendingRow(expireTime, dataRow.link()));
+                    cctx.ttl().onPendingEntryAdded(expireTime);
+                }
+            }
+            finally {
+                busyLock.leaveBusy();
+            }
         }
 
         /** {@inheritDoc} */
         @Override public void remove(KeyCacheObject key, int partId) throws IgniteCheckedException {
-            DataRow dataRow = dataTree.remove(new KeySearchRow(key.hashCode(), key, 0));
+            if (!busyLock.enterBusy())
+                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");
 
-            CacheObject val = null;
-            GridCacheVersion ver = null;
+            try {
+                DataRow dataRow = dataTree.remove(new KeySearchRow(key.hashCode(), key, 0));
 
-            if (dataRow != null) {
-                assert dataRow.link() != 0 : dataRow;
+                CacheObject val = null;
+                GridCacheVersion ver = null;
 
-                if (pendingEntries != null && dataRow.expireTime() != 0)
-                    pendingEntries.remove(new PendingRow(dataRow.expireTime(), dataRow.link()));
+                if (dataRow != null) {
+                    assert dataRow.link() != 0 : dataRow;
 
-                storageSize.decrementAndGet();
+                    if (pendingEntries != null && dataRow.expireTime() != 0)
+                        pendingEntries.remove(new PendingRow(dataRow.expireTime(), dataRow.link()));
 
-                val = dataRow.value();
+                    storageSize.decrementAndGet();
 
-                ver = dataRow.version();
-            }
+                    val = dataRow.value();
 
-            if (indexingEnabled) {
-                GridCacheQueryManager qryMgr = cctx.queries();
+                    ver = dataRow.version();
+                }
 
-                assert qryMgr.enabled();
+                if (indexingEnabled) {
+                    GridCacheQueryManager qryMgr = cctx.queries();
 
-                qryMgr.remove(key, partId, val, ver);
-            }
+                    assert qryMgr.enabled();
+
+                    qryMgr.remove(key, partId, val, ver);
+                }
 
-            if (dataRow != null)
-                rowStore.removeRow(dataRow.link());
+                if (dataRow != null)
+                    rowStore.removeRow(dataRow.link());
+            }
+            finally {
+                busyLock.leaveBusy();
+            }
         }
 
         /** {@inheritDoc} */

diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
@@ -2359,9 +2359,7 @@ private void onLeave(GridCacheGateway<K, V> gate) {
 
     /** {@inheritDoc} */
     @Override public IgniteFuture<?> rebalance() {
-        ctx.preloader().forcePreload();
-
-        return new IgniteFutureImpl<>(ctx.preloader().syncFuture());
+        return new IgniteFutureImpl<>(ctx.preloader().forceRebalance());
     }
 
     /** {@inheritDoc} */