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 2016/02/12 12:24:46 UTC
[1/2] ignite git commit: IGNITE-2509 - Fixed evictions for
OFFHEAP_VALUES memory mode - Fixes #452.
Repository: ignite
Updated Branches:
refs/heads/ignite-1.5.7 df9807061 -> 2e2d271d2
IGNITE-2509 - Fixed evictions for OFFHEAP_VALUES memory mode - Fixes #452.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f5261a00
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f5261a00
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f5261a00
Branch: refs/heads/ignite-1.5.7
Commit: f5261a00aff96c98e0c78bed7c3c82e810a4e5c6
Parents: df98070
Author: vershov <ve...@gridgain.com>
Authored: Mon Feb 8 21:54:48 2016 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Feb 12 13:16:47 2016 +0300
----------------------------------------------------------------------
.../cache/CacheEvictableEntryImpl.java | 4 +-
.../GridCacheOffHeapValuesEvictionSelfTest.java | 171 +++++++++++++++++++
.../ignite/testsuites/IgniteCacheTestSuite.java | 2 +
3 files changed, 175 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5261a00/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEvictableEntryImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEvictableEntryImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEvictableEntryImpl.java
index 7cea4f6..9f4d9d7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEvictableEntryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEvictableEntryImpl.java
@@ -103,7 +103,7 @@ public class CacheEvictableEntryImpl<K, V> implements EvictableEntry<K, V> {
byte[] valBytes = null;
- if (cctx.useOffheapEntry())
+ if (cctx.offheapTiered())
valBytes = cctx.offheap().get(cctx.swap().spaceName(), cached.partition(), key, keyBytes);
else {
CacheObject cacheObj = cached.valueBytes();
@@ -231,4 +231,4 @@ public class CacheEvictableEntryImpl<K, V> implements EvictableEntry<K, V> {
@Override public String toString() {
return S.toString(CacheEvictableEntryImpl.class, this);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5261a00/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java
new file mode 100644
index 0000000..9baab33
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.cache;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheMemoryMode;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.eviction.fifo.FifoEvictionPolicy;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.testframework.GridTestUtils;
+
+/**
+ * Test for eviction with offHeap.
+ */
+public class GridCacheOffHeapValuesEvictionSelfTest extends GridCacheAbstractSelfTest {
+
+ private static final int VAL_SIZE = 512 * 1024; // bytes
+ private static final int MAX_VALS_AMOUNT = 100;
+ private static final int MAX_MEMORY_SIZE = MAX_VALS_AMOUNT * VAL_SIZE;
+ private static final int VALS_AMOUNT = MAX_VALS_AMOUNT * 2;
+ private static final int THREAD_COUNT = 4;
+
+ @Override protected int gridCount() {
+ return 1;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPutOnHeap() throws Exception {
+ CacheConfiguration<Integer, Object> ccfg = cacheConfiguration(grid(0).name());
+ ccfg.setName("testPutOffHeapValues");
+ ccfg.setStatisticsEnabled(true);
+ ccfg.setOffHeapMaxMemory(MAX_MEMORY_SIZE);
+
+ FifoEvictionPolicy plc = new FifoEvictionPolicy();
+ plc.setMaxMemorySize(MAX_MEMORY_SIZE);
+
+ ccfg.setSwapEnabled(true);
+ ccfg.setMemoryMode(CacheMemoryMode.OFFHEAP_VALUES);
+
+ ccfg.setEvictionPolicy(plc);
+
+ final IgniteCache<Integer, Object> cache = grid(0).getOrCreateCache(ccfg);
+
+ fillCache(cache, getTestTimeout());
+
+ assertEquals(VALS_AMOUNT * THREAD_COUNT, cache.size(CachePeekMode.ALL));
+ assertEquals(0, cache.size(CachePeekMode.NEAR));
+ assertEquals(0, cache.size(CachePeekMode.OFFHEAP));
+ assertTrue(MAX_VALS_AMOUNT >= cache.size(CachePeekMode.ONHEAP));
+ assertTrue(MAX_VALS_AMOUNT - 5 <= cache.size(CachePeekMode.ONHEAP));
+ assertEquals(cache.size(CachePeekMode.ALL) - cache.size(CachePeekMode.ONHEAP), cache.size(CachePeekMode.SWAP));
+ }
+
+ /**
+ * swap disabled -> entries discarded
+ * @throws Exception If failed.
+ */
+ public void testPutOnHeapWithOffHeap() throws Exception {
+ final int PLC_MAX_SIZE = 50;
+
+ CacheConfiguration<Integer, Object> ccfg = cacheConfiguration(grid(0).name());
+ ccfg.setName("testPutOnHeapWithOffHeap");
+ ccfg.setStatisticsEnabled(true);
+ ccfg.setOffHeapMaxMemory(MAX_MEMORY_SIZE);
+
+ FifoEvictionPolicy plc = new FifoEvictionPolicy();
+ plc.setMaxMemorySize(MAX_MEMORY_SIZE);
+ plc.setMaxSize(PLC_MAX_SIZE);
+
+ ccfg.setSwapEnabled(false);
+ ccfg.setMemoryMode(CacheMemoryMode.ONHEAP_TIERED);
+
+ ccfg.setEvictionPolicy(plc);
+
+ final IgniteCache<Integer, Object> cache = grid(0).getOrCreateCache(ccfg);
+
+ fillCache(cache, getTestTimeout());
+
+ assertEquals(cache.size(CachePeekMode.ONHEAP) + cache.size(CachePeekMode.OFFHEAP), cache.size(CachePeekMode.ALL));
+ assertEquals(0, cache.size(CachePeekMode.NEAR));
+ assertEquals(0, cache.size(CachePeekMode.SWAP));
+ assertTrue(PLC_MAX_SIZE >= cache.size(CachePeekMode.ONHEAP));
+ assertTrue(PLC_MAX_SIZE - 5 <= cache.size(CachePeekMode.ONHEAP));
+ assertTrue(MAX_VALS_AMOUNT >= cache.size(CachePeekMode.OFFHEAP));
+ assertTrue(MAX_VALS_AMOUNT - 5 <= cache.size(CachePeekMode.OFFHEAP));
+ assertEquals(cache.size(CachePeekMode.ALL) - cache.size(CachePeekMode.ONHEAP) - cache.size(CachePeekMode.OFFHEAP),
+ cache.size(CachePeekMode.SWAP));
+
+ assertTrue((MAX_VALS_AMOUNT + 5) * VAL_SIZE > cache.metrics().getOffHeapAllocatedSize());
+ assertTrue((MAX_VALS_AMOUNT - 5) * VAL_SIZE < cache.metrics().getOffHeapAllocatedSize());
+ }
+
+ /**
+ * swap enabled -> entries are not discarded
+ * @throws Exception
+ */
+ public void testOnHeapWithOffHeapSwap() throws Exception{
+ final int PLC_MAX_SIZE = 50;
+
+ CacheConfiguration<Integer, Object> ccfg = cacheConfiguration(grid(0).name());
+ ccfg.setName("testOnHeapWithOffHeapSwap");
+ ccfg.setStatisticsEnabled(true);
+ ccfg.setOffHeapMaxMemory(MAX_MEMORY_SIZE);
+
+ FifoEvictionPolicy plc = new FifoEvictionPolicy();
+ plc.setMaxMemorySize(MAX_MEMORY_SIZE);
+ plc.setMaxSize(PLC_MAX_SIZE);
+
+ ccfg.setSwapEnabled(true);
+ ccfg.setMemoryMode(CacheMemoryMode.ONHEAP_TIERED);
+
+ ccfg.setEvictionPolicy(plc);
+
+ final IgniteCache<Integer, Object> cache = grid(0).getOrCreateCache(ccfg);
+
+ fillCache(cache, getTestTimeout());
+
+ assertEquals(cache.size(CachePeekMode.SWAP) + cache.size(CachePeekMode.ONHEAP) +
+ cache.size(CachePeekMode.OFFHEAP), cache.size(CachePeekMode.ALL));
+
+ assertTrue(PLC_MAX_SIZE >= cache.size(CachePeekMode.ONHEAP));
+ assertTrue(PLC_MAX_SIZE - 5 <= cache.size(CachePeekMode.ONHEAP));
+ assertTrue(MAX_VALS_AMOUNT >= cache.size(CachePeekMode.OFFHEAP));
+ assertTrue(MAX_VALS_AMOUNT - 5 <= cache.size(CachePeekMode.OFFHEAP));
+
+ assertTrue((MAX_VALS_AMOUNT + 5) * VAL_SIZE > cache.metrics().getOffHeapAllocatedSize());
+ assertTrue((MAX_VALS_AMOUNT - 5) * VAL_SIZE < cache.metrics().getOffHeapAllocatedSize());
+ }
+
+ private static void fillCache(final IgniteCache<Integer, Object> cache, long timeout) throws Exception{
+ final byte[] val = new byte[VAL_SIZE];
+ final AtomicInteger keyStart = new AtomicInteger(0);
+ final CountDownLatch latch = new CountDownLatch(4);
+
+ GridTestUtils.runMultiThreaded(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ final int start = keyStart.addAndGet(VALS_AMOUNT);
+
+ for (int i = start; i < start + VALS_AMOUNT; i++)
+ cache.put(i, val);
+
+ latch.countDown();
+
+ return null;
+ }
+ }, THREAD_COUNT, "test");
+
+ latch.await(timeout, TimeUnit.MILLISECONDS);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5261a00/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
index 88d0834..90125b1 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
@@ -65,6 +65,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheOffHeapTieredAtomicS
import org.apache.ignite.internal.processors.cache.GridCacheOffHeapTieredEvictionAtomicSelfTest;
import org.apache.ignite.internal.processors.cache.GridCacheOffHeapTieredEvictionSelfTest;
import org.apache.ignite.internal.processors.cache.GridCacheOffHeapTieredSelfTest;
+import org.apache.ignite.internal.processors.cache.GridCacheOffHeapValuesEvictionSelfTest;
import org.apache.ignite.internal.processors.cache.GridCacheP2PUndeploySelfTest;
import org.apache.ignite.internal.processors.cache.GridCachePartitionedLocalStoreSelfTest;
import org.apache.ignite.internal.processors.cache.GridCachePartitionedOffHeapLocalStoreSelfTest;
@@ -247,6 +248,7 @@ public class IgniteCacheTestSuite extends TestSuite {
suite.addTestSuite(GridCacheStorePutxSelfTest.class);
suite.addTestSuite(GridCacheOffHeapMultiThreadedUpdateSelfTest.class);
suite.addTestSuite(GridCacheOffHeapAtomicMultiThreadedUpdateSelfTest.class);
+ suite.addTestSuite(GridCacheOffHeapValuesEvictionSelfTest.class);
suite.addTestSuite(GridCacheColocatedTxStoreExceptionSelfTest.class);
suite.addTestSuite(GridCacheReplicatedTxStoreExceptionSelfTest.class);
suite.addTestSuite(GridCacheLocalTxStoreExceptionSelfTest.class);
[2/2] ignite git commit: IGNITE-2509 - Fixed offheap metrics - Fixes
#470.
Posted by ag...@apache.org.
IGNITE-2509 - Fixed offheap metrics - Fixes #470.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2e2d271d
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2e2d271d
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2e2d271d
Branch: refs/heads/ignite-1.5.7
Commit: 2e2d271d28328f3d483fb1570d2ce88e73b51707
Parents: f5261a0
Author: vershov <ve...@gridgain.com>
Authored: Fri Feb 12 12:51:50 2016 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Feb 12 13:21:59 2016 +0300
----------------------------------------------------------------------
.../org/apache/ignite/cache/CacheMemoryMode.java | 2 ++
.../processors/cache/GridCacheAdapter.java | 7 +++++++
.../processors/cache/GridCacheSwapManager.java | 3 ++-
.../GridCacheOffHeapValuesEvictionSelfTest.java | 17 +++++++++++++++--
4 files changed, 26 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2e2d271d/modules/core/src/main/java/org/apache/ignite/cache/CacheMemoryMode.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/CacheMemoryMode.java b/modules/core/src/main/java/org/apache/ignite/cache/CacheMemoryMode.java
index a596824..0133327 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/CacheMemoryMode.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/CacheMemoryMode.java
@@ -56,6 +56,8 @@ public enum CacheMemoryMode {
* Entry keys will be stored on heap memory, and values will be stored in offheap memory. Note
* that in this mode entries can be evicted only to swap. The evictions will happen according
* to configured {@link EvictionPolicy}.
+ * <p/>
+ * Size returned by {@link CachePeekMode#OFFHEAP} is always zero, for this mode.
*/
OFFHEAP_VALUES,
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/2e2d271d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
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
index 0f66550..37347ef 100644
--- 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
@@ -54,6 +54,7 @@ import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheInterceptor;
+import org.apache.ignite.cache.CacheMemoryMode;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.affinity.Affinity;
@@ -4167,6 +4168,12 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/** {@inheritDoc} */
@Override public long offHeapAllocatedSize() {
+ if (ctx.config().getMemoryMode() == CacheMemoryMode.OFFHEAP_VALUES) {
+ assert ctx.unsafeMemory() != null;
+
+ return ctx.unsafeMemory().allocatedSize();
+ }
+
GridCacheSwapManager swapMgr = ctx.swap();
return swapMgr != null ? swapMgr.offHeapAllocatedSize() : -1;
http://git-wip-us.apache.org/repos/asf/ignite/blob/2e2d271d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSwapManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSwapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSwapManager.java
index 37b5e15..cbf09bc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSwapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSwapManager.java
@@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentMap;
import javax.cache.Cache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheMemoryMode;
import org.apache.ignite.internal.managers.swapspace.GridSwapSpaceManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
@@ -85,7 +86,7 @@ public class GridCacheSwapManager extends GridCacheManagerAdapter {
/** Flag to indicate if swap is enabled. */
private boolean swapEnabled;
- /** Flag to indicate if offheap is enabled. */
+ /** Flag to indicate if offheap is enabled. {@link CacheMemoryMode#OFFHEAP_VALUES} treated as offheap disabled. */
private boolean offheapEnabled;
/** Swap listeners. */
http://git-wip-us.apache.org/repos/asf/ignite/blob/2e2d271d/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java
index 9baab33..0efd89b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapValuesEvictionSelfTest.java
@@ -33,12 +33,18 @@ import org.apache.ignite.testframework.GridTestUtils;
*/
public class GridCacheOffHeapValuesEvictionSelfTest extends GridCacheAbstractSelfTest {
+ /** */
private static final int VAL_SIZE = 512 * 1024; // bytes
+ /** */
private static final int MAX_VALS_AMOUNT = 100;
+ /** */
private static final int MAX_MEMORY_SIZE = MAX_VALS_AMOUNT * VAL_SIZE;
+ /** */
private static final int VALS_AMOUNT = MAX_VALS_AMOUNT * 2;
+ /** */
private static final int THREAD_COUNT = 4;
+ /** {@inheritDoc} */
@Override protected int gridCount() {
return 1;
}
@@ -46,7 +52,7 @@ public class GridCacheOffHeapValuesEvictionSelfTest extends GridCacheAbstractSel
/**
* @throws Exception If failed.
*/
- public void testPutOnHeap() throws Exception {
+ public void testPutValuesOffHeap() throws Exception {
CacheConfiguration<Integer, Object> ccfg = cacheConfiguration(grid(0).name());
ccfg.setName("testPutOffHeapValues");
ccfg.setStatisticsEnabled(true);
@@ -70,6 +76,10 @@ public class GridCacheOffHeapValuesEvictionSelfTest extends GridCacheAbstractSel
assertTrue(MAX_VALS_AMOUNT >= cache.size(CachePeekMode.ONHEAP));
assertTrue(MAX_VALS_AMOUNT - 5 <= cache.size(CachePeekMode.ONHEAP));
assertEquals(cache.size(CachePeekMode.ALL) - cache.size(CachePeekMode.ONHEAP), cache.size(CachePeekMode.SWAP));
+
+ assertTrue((MAX_VALS_AMOUNT + 5) * VAL_SIZE > cache.metrics().getOffHeapAllocatedSize());
+ assertTrue((MAX_VALS_AMOUNT - 5) * VAL_SIZE < cache.metrics().getOffHeapAllocatedSize());
+ assertTrue(cache.metrics().getOffHeapAllocatedSize() >= cache.size(CachePeekMode.ONHEAP) * VAL_SIZE);
}
/**
@@ -109,6 +119,7 @@ public class GridCacheOffHeapValuesEvictionSelfTest extends GridCacheAbstractSel
assertTrue((MAX_VALS_AMOUNT + 5) * VAL_SIZE > cache.metrics().getOffHeapAllocatedSize());
assertTrue((MAX_VALS_AMOUNT - 5) * VAL_SIZE < cache.metrics().getOffHeapAllocatedSize());
+ assertTrue(cache.metrics().getOffHeapAllocatedSize() >= cache.size(CachePeekMode.OFFHEAP) * VAL_SIZE);
}
/**
@@ -146,12 +157,14 @@ public class GridCacheOffHeapValuesEvictionSelfTest extends GridCacheAbstractSel
assertTrue((MAX_VALS_AMOUNT + 5) * VAL_SIZE > cache.metrics().getOffHeapAllocatedSize());
assertTrue((MAX_VALS_AMOUNT - 5) * VAL_SIZE < cache.metrics().getOffHeapAllocatedSize());
+ assertTrue(cache.metrics().getOffHeapAllocatedSize() >= cache.size(CachePeekMode.OFFHEAP) * VAL_SIZE);
}
+ /** Fill cache with values. */
private static void fillCache(final IgniteCache<Integer, Object> cache, long timeout) throws Exception{
final byte[] val = new byte[VAL_SIZE];
final AtomicInteger keyStart = new AtomicInteger(0);
- final CountDownLatch latch = new CountDownLatch(4);
+ final CountDownLatch latch = new CountDownLatch(THREAD_COUNT);
GridTestUtils.runMultiThreaded(new Callable<Void>() {
@Override public Void call() throws Exception {