| /* |
| * 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.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import javax.cache.Cache; |
| import org.apache.ignite.Ignite; |
| import org.apache.ignite.IgniteCache; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.cache.CachePeekMode; |
| import org.apache.ignite.cache.affinity.Affinity; |
| import org.apache.ignite.cache.eviction.fifo.FifoEvictionPolicy; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.configuration.NearCacheConfiguration; |
| import org.apache.ignite.internal.IgniteEx; |
| import org.apache.ignite.internal.IgniteKernal; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.util.GridEmptyCloseableIterator; |
| import org.apache.ignite.internal.util.typedef.T2; |
| import org.apache.ignite.spi.IgniteSpiCloseableIterator; |
| import org.junit.Test; |
| import static org.apache.ignite.cache.CacheMode.PARTITIONED; |
| import static org.apache.ignite.cache.CacheMode.REPLICATED; |
| import static org.apache.ignite.cache.CachePeekMode.ALL; |
| import static org.apache.ignite.cache.CachePeekMode.BACKUP; |
| import static org.apache.ignite.cache.CachePeekMode.NEAR; |
| import static org.apache.ignite.cache.CachePeekMode.OFFHEAP; |
| import static org.apache.ignite.cache.CachePeekMode.ONHEAP; |
| import static org.apache.ignite.cache.CachePeekMode.PRIMARY; |
| |
| /** |
| * Tests for methods using {@link CachePeekMode}: |
| * <ul> |
| * <li>{@link IgniteCache#localPeek(Object, CachePeekMode...)}</li> |
| * <li>{@link IgniteCache#localSize(CachePeekMode...)}</li> |
| * <li>{@link IgniteCache#size(CachePeekMode...)}</li> |
| * <li>{@link IgniteCache#localEntries(CachePeekMode...)}</li> |
| * </ul> |
| */ |
| public abstract class IgniteCachePeekModesAbstractTest extends IgniteCacheAbstractTest { |
| /** */ |
| private static final int HEAP_ENTRIES = 30; |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { |
| IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); |
| |
| return cfg; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected NearCacheConfiguration nearConfiguration() { |
| return null; |
| } |
| |
| /** |
| * @return Has near cache flag. |
| */ |
| protected boolean hasNearCache() { |
| return false; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected CacheConfiguration cacheConfiguration(String igniteInstanceName) throws Exception { |
| CacheConfiguration ccfg = super.cacheConfiguration(igniteInstanceName); |
| |
| ccfg.setBackups(1); |
| |
| if (hasNearCache()) |
| ccfg.setNearConfiguration(new NearCacheConfiguration()); |
| |
| FifoEvictionPolicy plc = new FifoEvictionPolicy(); |
| plc.setMaxSize(HEAP_ENTRIES); |
| |
| ccfg.setEvictionPolicy(plc); |
| ccfg.setOnheapCacheEnabled(true); |
| |
| return ccfg; |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testLocalPeek() throws Exception { |
| checkAffinityPeek(0); |
| |
| checkAffinityPeek(1); |
| |
| checkStorage(0); |
| |
| checkStorage(1); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkAffinityPeek(int nodeIdx) throws Exception { |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| final String val = "1"; |
| |
| Integer key = null; |
| |
| try { |
| if (cacheMode() == REPLICATED) { |
| key = backupKey(cache0); |
| |
| cache0.put(key, val); |
| |
| assertEquals(val, cache0.localPeek(key, ALL)); |
| assertEquals(val, cache0.localPeek(key, BACKUP)); |
| assertNull(cache0.localPeek(key, NEAR)); |
| assertNull(cache0.localPeek(key, PRIMARY)); |
| } |
| else { |
| key = nearKey(cache0); |
| |
| cache0.put(key, val); |
| |
| if (hasNearCache()) { |
| assertEquals(val, cache0.localPeek(key, NEAR)); |
| assertEquals(val, cache0.localPeek(key, ALL)); |
| } |
| else { |
| assertNull(cache0.localPeek(key, NEAR)); |
| assertNull(cache0.localPeek(key, ALL)); |
| } |
| |
| assertNull(cache0.localPeek(key, PRIMARY)); |
| assertNull(cache0.localPeek(key, BACKUP)); |
| } |
| |
| Affinity<Integer> aff = ignite(0).affinity(DEFAULT_CACHE_NAME); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| if (i == nodeIdx) |
| continue; |
| |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertNull(cache.localPeek(key, NEAR)); |
| |
| if (aff.isPrimary(ignite(i).cluster().localNode(), key)) { |
| assertEquals(val, cache.localPeek(key, PRIMARY)); |
| assertEquals(val, cache.localPeek(key, ALL)); |
| assertNull(cache.localPeek(key, BACKUP)); |
| assertNull(cache.localPeek(key, NEAR)); |
| } |
| else if (aff.isBackup(ignite(i).cluster().localNode(), key)) { |
| assertEquals(val, cache.localPeek(key, BACKUP)); |
| assertEquals(val, cache.localPeek(key, ALL)); |
| assertNull(cache.localPeek(key, PRIMARY)); |
| assertNull(cache.localPeek(key, NEAR)); |
| } |
| else { |
| assertNull(cache.localPeek(key, ALL)); |
| assertNull(cache.localPeek(key, PRIMARY)); |
| assertNull(cache.localPeek(key, BACKUP)); |
| assertNull(cache.localPeek(key, NEAR)); |
| } |
| } |
| } |
| finally { |
| if (key != null) |
| cache0.remove(key); |
| } |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkStorage(int nodeIdx) throws Exception { |
| if (true) // TODO GG-11148. |
| return; |
| |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| List<Integer> keys = primaryKeys(cache0, 100, 10_000); |
| |
| try { |
| final String val = "test_value"; |
| |
| for (Integer key : keys) |
| cache0.put(key, val); |
| |
| Ignite ignite = ignite(nodeIdx); |
| |
| GridCacheAdapter<Integer, String> internalCache = |
| ((IgniteKernal)ignite).context().cache().internalCache(DEFAULT_CACHE_NAME); |
| |
| CacheObjectContext coctx = internalCache.context().cacheObjectContext(); |
| |
| Set<Integer> swapKeys = new HashSet<>(); |
| |
| // TODO GG-11148. |
| // SwapSpaceSpi swap = ignite.configuration().getSwapSpaceSpi(); |
| // |
| // IgniteSpiCloseableIterator<KeyCacheObject> it = swap.keyIterator(SPACE_NAME, null); |
| |
| IgniteSpiCloseableIterator<KeyCacheObject> it = new GridEmptyCloseableIterator<>(); |
| |
| assertNotNull(it); |
| |
| while (it.hasNext()) { |
| KeyCacheObject next = it.next(); |
| |
| assertTrue(swapKeys.add((Integer)next.value(coctx, false))); |
| } |
| |
| assertFalse(swapKeys.isEmpty()); |
| |
| assertTrue(swapKeys.size() + HEAP_ENTRIES < 100); |
| |
| Set<Integer> offheapKeys = new HashSet<>(); |
| |
| // TODO GG-11148. |
| Iterator<Map.Entry<Integer, String>> offheapIt = Collections.EMPTY_MAP.entrySet().iterator(); |
| |
| // if (internalCache.context().isNear()) |
| // offheapIt = internalCache.context().near().dht().context().swap().lazyOffHeapIterator(false); |
| // else |
| // offheapIt = internalCache.context().swap().lazyOffHeapIterator(false); |
| |
| while (offheapIt.hasNext()) { |
| Map.Entry<Integer, String> e = offheapIt.next(); |
| |
| assertTrue(offheapKeys.add(e.getKey())); |
| |
| assertFalse(swapKeys.contains(e.getKey())); |
| } |
| |
| assertFalse(offheapKeys.isEmpty()); |
| |
| Set<Integer> heapKeys = new HashSet<>(keys); |
| |
| heapKeys.removeAll(offheapKeys); |
| heapKeys.removeAll(swapKeys); |
| |
| assertFalse(heapKeys.isEmpty()); |
| |
| log.info("Keys [swap=" + swapKeys.size() + |
| ", offheap=" + offheapKeys.size() + |
| ", heap=" + heapKeys.size() + ']'); |
| |
| assertEquals(100, swapKeys.size() + offheapKeys.size() + heapKeys.size()); |
| |
| for (Integer key : swapKeys) { |
| assertEquals(val, cache0.localPeek(key)); |
| assertEquals(val, cache0.localPeek(key, PRIMARY)); |
| assertEquals(val, cache0.localPeek(key, ONHEAP)); |
| assertEquals(val, cache0.localPeek(key, ONHEAP, OFFHEAP)); |
| assertEquals(val, cache0.localPeek(key, PRIMARY, ONHEAP)); |
| assertEquals(val, cache0.localPeek(key, PRIMARY, ONHEAP, OFFHEAP)); |
| |
| assertNull(cache0.localPeek(key, BACKUP)); |
| assertNull(cache0.localPeek(key, NEAR)); |
| |
| assertNull(cache0.localPeek(key, ONHEAP)); |
| assertNull(cache0.localPeek(key, OFFHEAP)); |
| } |
| |
| for (Integer key : offheapKeys) { |
| assertEquals(val, cache0.localPeek(key, OFFHEAP)); |
| assertEquals(val, cache0.localPeek(key, ONHEAP, OFFHEAP)); |
| assertEquals(val, cache0.localPeek(key, ONHEAP, OFFHEAP)); |
| assertEquals(val, cache0.localPeek(key, PRIMARY, OFFHEAP)); |
| |
| assertNull(cache0.localPeek(key, OFFHEAP, BACKUP)); |
| assertNull(cache0.localPeek(key, OFFHEAP, NEAR)); |
| |
| assertNull(cache0.localPeek(key, ONHEAP)); |
| assertNull(cache0.localPeek(key)); |
| } |
| |
| for (Integer key : heapKeys) { |
| assertEquals(val, cache0.localPeek(key, ONHEAP)); |
| assertEquals(val, cache0.localPeek(key, ONHEAP)); |
| assertEquals(val, cache0.localPeek(key, OFFHEAP, ONHEAP)); |
| assertEquals(val, cache0.localPeek(key, PRIMARY, ONHEAP)); |
| |
| assertNull(cache0.localPeek(key, ONHEAP, BACKUP)); |
| assertNull(cache0.localPeek(key, ONHEAP, NEAR)); |
| |
| assertNull(cache0.localPeek(key)); |
| assertNull(cache0.localPeek(key, OFFHEAP)); |
| } |
| } |
| finally { |
| cache0.removeAll(new HashSet<>(keys)); |
| } |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkSizeAffinityFilter(int nodeIdx) throws Exception { |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| final int PUT_KEYS = 10; |
| |
| List<Integer> keys = null; |
| |
| try { |
| if (cacheMode() == REPLICATED) { |
| keys = backupKeys(cache0, 10, 0); |
| |
| for (Integer key : keys) |
| cache0.put(key, String.valueOf(key)); |
| |
| assertEquals(PUT_KEYS, cache0.localSize(BACKUP)); |
| assertEquals(PUT_KEYS, cache0.localSize(ALL)); |
| assertEquals(0, cache0.localSize()); |
| assertEquals(0, cache0.localSize(PRIMARY)); |
| assertEquals(0, cache0.localSize(NEAR)); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(0, cache.size(NEAR)); |
| assertEquals(PUT_KEYS, cache.size(PRIMARY)); |
| assertEquals(PUT_KEYS * (gridCount() - 1), cache.size(BACKUP)); |
| assertEquals(PUT_KEYS * gridCount(), cache.size(PRIMARY, BACKUP)); |
| assertEquals(PUT_KEYS * gridCount(), cache.size(ALL)); // Primary + backups. |
| } |
| } |
| else { |
| keys = nearKeys(cache0, PUT_KEYS, 0); |
| |
| for (Integer key : keys) |
| cache0.put(key, String.valueOf(key)); |
| |
| if (hasNearCache()) { |
| assertEquals(0, cache0.localSize()); |
| assertEquals(PUT_KEYS, cache0.localSize(ALL)); |
| assertEquals(PUT_KEYS, cache0.localSize(NEAR)); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(PUT_KEYS, cache.size(NEAR)); |
| assertEquals(PUT_KEYS, cache.size(BACKUP)); |
| assertEquals(PUT_KEYS * 2, cache.size(PRIMARY, BACKUP)); |
| assertEquals(PUT_KEYS * 2 + PUT_KEYS, cache.size(ALL)); // Primary + backups + near. |
| } |
| } |
| else { |
| assertEquals(0, cache0.localSize()); |
| assertEquals(0, cache0.localSize(ALL)); |
| assertEquals(0, cache0.localSize(NEAR)); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(0, cache.size(NEAR)); |
| assertEquals(PUT_KEYS, cache.size(BACKUP)); |
| assertEquals(PUT_KEYS * 2, cache.size(PRIMARY, BACKUP)); |
| assertEquals(PUT_KEYS * 2, cache.size(ALL)); // Primary + backups. |
| } |
| } |
| |
| assertEquals(0, cache0.localSize(BACKUP)); |
| assertEquals(0, cache0.localSize(PRIMARY)); |
| } |
| |
| checkPrimarySize(PUT_KEYS); |
| |
| Affinity<Integer> aff = ignite(0).affinity(DEFAULT_CACHE_NAME); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| if (i == nodeIdx) |
| continue; |
| |
| ClusterNode node = ignite(i).cluster().localNode(); |
| |
| int primary = 0; |
| int backups = 0; |
| |
| for (Integer key : keys) { |
| if (aff.isPrimary(node, key)) |
| primary++; |
| else if (aff.isBackup(node, key)) |
| backups++; |
| } |
| |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(primary, cache.localSize(PRIMARY)); |
| assertEquals(backups, cache.localSize(BACKUP)); |
| assertEquals(primary + backups, cache.localSize(PRIMARY, BACKUP)); |
| assertEquals(primary + backups, cache.localSize(BACKUP, PRIMARY)); |
| assertEquals(primary + backups, cache.localSize(ALL)); |
| } |
| |
| cache0.remove(keys.get(0)); |
| |
| checkPrimarySize(PUT_KEYS - 1); |
| |
| if (cacheMode() == REPLICATED) { |
| assertEquals(PUT_KEYS - 1, cache0.localSize(ALL)); |
| assertEquals(0, cache0.localSize(PRIMARY)); |
| assertEquals(PUT_KEYS - 1, cache0.localSize(BACKUP)); |
| } |
| else { |
| if (hasNearCache()) |
| assertEquals(PUT_KEYS - 1, cache0.localSize(ALL)); |
| else |
| assertEquals(0, cache0.localSize(ALL)); |
| } |
| } |
| finally { |
| if (keys != null) |
| cache0.removeAll(new HashSet<>(keys)); |
| } |
| |
| checkEmpty(); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkPartitionSizeAffinityFilter(int nodeIdx) throws Exception { |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| final int PUT_KEYS = 10; |
| |
| int part = nodeIdx; |
| |
| List<Integer> keys = null; |
| |
| try { |
| if (cacheMode() == REPLICATED) { |
| keys = backupKeys(cache0, 10, 0); |
| |
| for (Integer key : keys) |
| cache0.put(key, String.valueOf(key)); |
| |
| int partSize = 0; |
| |
| for (Integer key : keys) { |
| int keyPart = ignite(nodeIdx).affinity(DEFAULT_CACHE_NAME).partition(key); |
| if (keyPart == part) |
| partSize++; |
| } |
| |
| assertEquals(PUT_KEYS, cache0.localSize(BACKUP)); |
| assertEquals(PUT_KEYS, cache0.localSize(ALL)); |
| assertEquals(partSize, cache0.localSizeLong(part, BACKUP)); |
| assertEquals(partSize, cache0.localSizeLong(part, ALL)); |
| assertEquals(0, cache0.localSizeLong(part, PRIMARY)); |
| assertEquals(0, cache0.localSizeLong(part, NEAR)); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| assertEquals(0, cache.size(NEAR)); |
| assertEquals(partSize, cache.sizeLong(part, PRIMARY)); |
| assertEquals(partSize * (gridCount() - 1), cache.sizeLong(part, BACKUP)); |
| assertEquals(partSize * gridCount(), cache.sizeLong(part, PRIMARY, BACKUP)); |
| assertEquals(partSize * gridCount(), cache.sizeLong(part, ALL)); // Primary + backups. |
| } |
| } |
| else { |
| keys = nearKeys(cache0, PUT_KEYS, 0); |
| |
| for (Integer key : keys) |
| cache0.put(key, String.valueOf(key)); |
| |
| int partSize = 0; |
| |
| for (Integer key :keys) { |
| int keyPart = ignite(nodeIdx).affinity(DEFAULT_CACHE_NAME).partition(key); |
| if (keyPart == part) |
| partSize++; |
| } |
| |
| if (hasNearCache()) { |
| assertEquals(0, cache0.localSize()); |
| assertEquals(0, cache0.localSizeLong(part, ALL)); |
| assertEquals(0, cache0.localSizeLong(part, NEAR)); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(0, cache.sizeLong(part, NEAR)); |
| assertEquals(partSize, cache.sizeLong(part, BACKUP)); |
| assertEquals(partSize * 2, cache.sizeLong(part, PRIMARY, BACKUP)); |
| assertEquals(partSize * 2, cache.sizeLong(part, ALL)); // Primary + backups + near. |
| } |
| } |
| else { |
| assertEquals(0, cache0.localSize()); |
| //assertEquals(partitionSize, cache0.localSizeLong(partition, ALL)); |
| assertEquals(0, cache0.localSizeLong(part, NEAR)); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(0, cache.size(NEAR)); |
| assertEquals(partSize, cache.sizeLong(part, BACKUP)); |
| assertEquals(partSize * 2, cache.sizeLong(part, PRIMARY, BACKUP)); |
| assertEquals(partSize * 2, cache.sizeLong(part, ALL)); // Primary + backups. |
| } |
| } |
| |
| assertEquals(0, cache0.localSize(BACKUP)); |
| assertEquals(0, cache0.localSize(PRIMARY)); |
| } |
| |
| checkPrimarySize(PUT_KEYS); |
| |
| Affinity<Integer> aff = ignite(0).affinity(DEFAULT_CACHE_NAME); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| if (i == nodeIdx) |
| continue; |
| |
| ClusterNode node = ignite(i).cluster().localNode(); |
| |
| int primary = 0; |
| int backups = 0; |
| |
| for (Integer key : keys) { |
| if (aff.isPrimary(node, key) && aff.partition(key) == part) |
| primary++; |
| else if (aff.isBackup(node, key) && aff.partition(key) == part) |
| backups++; |
| } |
| |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(primary, cache.localSizeLong(part, PRIMARY)); |
| assertEquals(backups, cache.localSizeLong(part, BACKUP)); |
| assertEquals(primary + backups, cache.localSizeLong(part, PRIMARY, BACKUP)); |
| assertEquals(primary + backups, cache.localSizeLong(part, BACKUP, PRIMARY)); |
| assertEquals(primary + backups, cache.localSizeLong(part, ALL)); |
| } |
| |
| cache0.remove(keys.get(0)); |
| |
| keys.remove(0); |
| |
| checkPrimarySize(PUT_KEYS - 1); |
| |
| int primary = 0; |
| int backups = 0; |
| |
| ClusterNode node = ignite(nodeIdx).cluster().localNode(); |
| |
| for (Integer key : keys) { |
| if (aff.isPrimary(node, key) && aff.partition(key) == part) |
| primary++; |
| else if (aff.isBackup(node, key) && aff.partition(key) == part) |
| backups++; |
| } |
| |
| if (cacheMode() == REPLICATED) { |
| assertEquals(primary + backups, cache0.localSizeLong(part, ALL)); |
| assertEquals(primary, cache0.localSizeLong(part, PRIMARY)); |
| assertEquals(backups, cache0.localSizeLong(part, BACKUP)); |
| } |
| else { |
| if (hasNearCache()) |
| assertEquals(0, cache0.localSizeLong(part, ALL)); |
| else |
| assertEquals(0, cache0.localSizeLong(part, ALL)); |
| } |
| } |
| finally { |
| if (keys != null) |
| cache0.removeAll(new HashSet<>(keys)); |
| } |
| |
| checkEmpty(); |
| } |
| |
| /** |
| * Checks size is zero. |
| */ |
| private void checkEmpty() { |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(0, cache.localSize()); |
| |
| assertEquals(0, cache.size()); |
| |
| for (CachePeekMode peekMode : CachePeekMode.values()) { |
| assertEquals(0, cache.localSize(peekMode)); |
| |
| assertEquals(0, cache.size(peekMode)); |
| } |
| } |
| |
| checkPrimarySize(0); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @return Tuple with primary and backup keys. |
| */ |
| private T2<List<Integer>, List<Integer>> swapKeys(int nodeIdx) { |
| // TODO: GG-11148. |
| // SwapSpaceSpi swap = ignite(nodeIdx).configuration().getSwapSpaceSpi(); |
| // |
| // IgniteSpiCloseableIterator<KeyCacheObject> it = swap.keyIterator(SPACE_NAME, null); |
| IgniteSpiCloseableIterator<KeyCacheObject> it = new GridEmptyCloseableIterator<>(); |
| |
| assertNotNull(it); |
| |
| Affinity aff = ignite(nodeIdx).affinity(DEFAULT_CACHE_NAME); |
| |
| ClusterNode node = ignite(nodeIdx).cluster().localNode(); |
| |
| List<Integer> primary = new ArrayList<>(); |
| List<Integer> backups = new ArrayList<>(); |
| |
| CacheObjectContext coctx = ((IgniteEx)ignite(nodeIdx)).context().cache().internalCache(DEFAULT_CACHE_NAME) |
| .context().cacheObjectContext(); |
| |
| while (it.hasNext()) { |
| Integer key = it.next().value(coctx, false); |
| |
| if (aff.isPrimary(node, key)) |
| primary.add(key); |
| else { |
| assertTrue(aff.isBackup(node, key)); |
| |
| backups.add(key); |
| } |
| } |
| |
| return new T2<>(primary, backups); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @return Tuple with number of primary and backup keys. |
| */ |
| private T2<Integer, Integer> swapKeysCount(int nodeIdx) { |
| T2<List<Integer>, List<Integer>> keys = swapKeys(nodeIdx); |
| |
| return new T2<>(keys.get1().size(), keys.get2().size()); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @return Tuple with primary and backup keys. |
| */ |
| private T2<List<Integer>, List<Integer>> offheapKeys(int nodeIdx) { |
| GridCacheAdapter<Integer, String> internalCache = |
| ((IgniteKernal)ignite(nodeIdx)).context().cache().internalCache(DEFAULT_CACHE_NAME); |
| |
| // TODO GG-11148. |
| Iterator<Map.Entry<Integer, String>> offheapIt = Collections.EMPTY_MAP.entrySet().iterator(); |
| // if (internalCache.context().isNear()) |
| // offheapIt = internalCache.context().near().dht().context().swap().lazyOffHeapIterator(false); |
| // else |
| // offheapIt = internalCache.context().swap().lazyOffHeapIterator(false); |
| |
| Affinity aff = ignite(nodeIdx).affinity(DEFAULT_CACHE_NAME); |
| |
| ClusterNode node = ignite(nodeIdx).cluster().localNode(); |
| |
| List<Integer> primary = new ArrayList<>(); |
| List<Integer> backups = new ArrayList<>(); |
| |
| while (offheapIt.hasNext()) { |
| Map.Entry<Integer, String> e = offheapIt.next(); |
| |
| if (aff.isPrimary(node, e.getKey())) |
| primary.add(e.getKey()); |
| else { |
| assertTrue(aff.isBackup(node, e.getKey())); |
| |
| backups.add(e.getKey()); |
| } |
| } |
| |
| return new T2<>(primary, backups); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @return Tuple with number of primary and backup keys. |
| */ |
| private T2<Integer, Integer> offheapKeysCount(int nodeIdx) { |
| T2<List<Integer>, List<Integer>> keys = offheapKeys(nodeIdx); |
| |
| return new T2<>(keys.get1().size(), keys.get2().size()); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @param part Cache partition. |
| * @return Tuple with number of primary and backup keys (one or both will be zero). |
| */ |
| private T2<Integer, Integer> offheapKeysCount(int nodeIdx, int part) throws IgniteCheckedException { |
| GridCacheContext ctx = ((IgniteEx)ignite(nodeIdx)).context().cache().internalCache(DEFAULT_CACHE_NAME).context(); |
| // Swap and offheap are disabled for near cache. |
| IgniteCacheOffheapManager offheapManager = ctx.isNear() ? ctx.near().dht().context().offheap() : ctx.offheap(); |
| //First count entries... |
| int cnt = (int)offheapManager.cacheEntriesCount(ctx.cacheId(), part); |
| |
| GridCacheAffinityManager affinity = ctx.affinity(); |
| AffinityTopologyVersion topVer = affinity.affinityTopologyVersion(); |
| |
| //And then find out whether they are primary or backup ones. |
| int primaryCnt = 0; |
| int backupCnt = 0; |
| if (affinity.primaryByPartition(ctx.localNode(), part, topVer)) |
| primaryCnt = cnt; |
| else if (affinity.backupByPartition(ctx.localNode(), part, topVer)) |
| backupCnt = cnt; |
| return new T2<>(primaryCnt, backupCnt); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkSizeStorageFilter(int nodeIdx) throws Exception { |
| if (true) // TODO GG-11148. |
| return; |
| |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| List<Integer> primaryKeys = primaryKeys(cache0, 100, 10_000); |
| List<Integer> backupKeys = backupKeys(cache0, 100, 10_000); |
| |
| try { |
| final String val = "test_value"; |
| |
| for (int i = 0; i < 100; i++) { |
| cache0.put(primaryKeys.get(i), val); |
| cache0.put(backupKeys.get(i), val); |
| } |
| |
| int totalKeys = 200; |
| |
| T2<Integer, Integer> swapKeys = swapKeysCount(nodeIdx); |
| |
| assertTrue(swapKeys.get1() > 0); |
| assertTrue(swapKeys.get2() > 0); |
| |
| T2<Integer, Integer> offheapKeys = offheapKeysCount(nodeIdx); |
| |
| assertTrue(offheapKeys.get1() > 0); |
| assertTrue(offheapKeys.get2() > 0); |
| |
| int totalSwap = swapKeys.get1() + swapKeys.get2(); |
| int totalOffheap = offheapKeys.get1() + offheapKeys.get2(); |
| |
| log.info("Local keys [total=" + totalKeys + ", offheap=" + offheapKeys + ", swap=" + swapKeys + ']'); |
| |
| assertTrue(totalSwap + totalOffheap < totalKeys); |
| |
| assertEquals(primaryKeys.size(), cache0.localSize()); |
| assertEquals(totalKeys, cache0.localSize(ALL)); |
| assertEquals(totalOffheap, cache0.localSize(PRIMARY, BACKUP, NEAR, OFFHEAP)); |
| assertEquals(totalSwap, cache0.localSize(PRIMARY, BACKUP, NEAR)); |
| assertEquals(totalKeys - (totalOffheap + totalSwap), cache0.localSize(PRIMARY, BACKUP, NEAR, ONHEAP)); |
| assertEquals(totalKeys, cache0.localSize(PRIMARY, BACKUP, NEAR, OFFHEAP, ONHEAP)); |
| |
| assertEquals(swapKeys.get1(), (Integer)cache0.localSize(PRIMARY)); |
| assertEquals(swapKeys.get2(), (Integer)cache0.localSize(BACKUP)); |
| |
| assertEquals(offheapKeys.get1(), (Integer)cache0.localSize(OFFHEAP, PRIMARY)); |
| assertEquals(offheapKeys.get2(), (Integer)cache0.localSize(OFFHEAP, BACKUP)); |
| |
| assertEquals(swapKeys.get1() + offheapKeys.get1(), cache0.localSize(OFFHEAP, PRIMARY)); |
| assertEquals(swapKeys.get2() + offheapKeys.get2(), cache0.localSize(OFFHEAP, BACKUP)); |
| |
| assertEquals(totalSwap + totalOffheap, cache0.localSize(PRIMARY, BACKUP, NEAR, OFFHEAP)); |
| |
| int globalSwapPrimary = 0; |
| int globalSwapBackup = 0; |
| |
| int globalOffheapPrimary = 0; |
| int globalOffheapBackup = 0; |
| |
| for (int i = 0; i < gridCount(); i++) { |
| T2<Integer, Integer> swap = swapKeysCount(i); |
| |
| globalSwapPrimary += swap.get1(); |
| globalSwapBackup += swap.get2(); |
| |
| T2<Integer, Integer> offheap = offheapKeysCount(i); |
| |
| globalOffheapPrimary += offheap.get1(); |
| globalOffheapBackup += offheap.get2(); |
| } |
| |
| int backups; |
| |
| if (cacheMode() == PARTITIONED) |
| backups = 1; |
| else // REPLICATED. |
| backups = gridCount() - 1; |
| |
| int globalTotal = totalKeys + totalKeys * backups; |
| int globalTotalSwap = globalSwapPrimary + globalSwapBackup; |
| int globalTotalOffheap = globalOffheapPrimary + globalOffheapBackup; |
| |
| log.info("Global keys [total=" + globalTotal + |
| ", offheap=" + globalTotalOffheap + |
| ", swap=" + globalTotalSwap + ']'); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(totalKeys, cache.size(PRIMARY)); |
| assertEquals(globalTotal, cache.size(ALL)); |
| assertEquals(globalTotal, cache.size(PRIMARY, BACKUP, NEAR, ONHEAP, OFFHEAP)); |
| assertEquals(globalTotal, cache.size(ONHEAP, OFFHEAP, PRIMARY, BACKUP)); |
| |
| assertEquals(globalTotalSwap, cache.size(PRIMARY, BACKUP, NEAR)); |
| assertEquals(globalSwapPrimary, cache.size(PRIMARY)); |
| assertEquals(globalSwapBackup, cache.size(BACKUP)); |
| |
| assertEquals(globalTotalOffheap, cache.size(PRIMARY, BACKUP, NEAR, OFFHEAP)); |
| assertEquals(globalOffheapPrimary, cache.size(OFFHEAP, PRIMARY)); |
| assertEquals(globalOffheapBackup, cache.size(OFFHEAP, BACKUP)); |
| |
| assertEquals(globalTotalSwap + globalTotalOffheap, cache.size(PRIMARY, BACKUP, NEAR, OFFHEAP)); |
| assertEquals(globalSwapPrimary + globalOffheapPrimary, cache.size(OFFHEAP, PRIMARY)); |
| assertEquals(globalSwapBackup + globalOffheapBackup, cache.size(OFFHEAP, BACKUP)); |
| |
| assertEquals(globalTotal - (globalTotalOffheap + globalTotalSwap), cache.size(PRIMARY, BACKUP, NEAR, ONHEAP)); |
| } |
| } |
| finally { |
| cache0.removeAll(new HashSet<>(primaryKeys)); |
| cache0.removeAll(new HashSet<>(backupKeys)); |
| } |
| |
| checkEmpty(); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkPartitionSizeStorageFilter(int nodeIdx) throws Exception { |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| int part = nodeIdx; |
| |
| List<Integer> primaryKeys = primaryKeys(cache0, 100, 10_000); |
| List<Integer> backupKeys = backupKeys(cache0, 100, 10_000); |
| |
| try { |
| final String val = "test_value"; |
| |
| for (int i = 0; i < 100; i++) { |
| cache0.put(primaryKeys.get(i), val); |
| cache0.put(backupKeys.get(i), val); |
| } |
| |
| int totalKeys = 200; |
| |
| T2<Integer, Integer> offheapKeys = offheapKeysCount(nodeIdx, part); |
| |
| int totalOffheap = offheapKeys.get1() + offheapKeys.get2(); |
| |
| log.info("Local keys [total=" + totalKeys + ", offheap=" + offheapKeys + ']'); |
| |
| assertTrue(totalOffheap < totalKeys); |
| |
| assertEquals(primaryKeys.size(), cache0.localSize()); |
| assertEquals(totalKeys, cache0.localSize(ALL)); |
| assertEquals(totalOffheap, cache0.localSizeLong(part, PRIMARY, BACKUP, NEAR, OFFHEAP)); |
| |
| assertEquals((long)offheapKeys.get1(), cache0.localSizeLong(part, OFFHEAP, PRIMARY)); |
| assertEquals((long)offheapKeys.get2(), cache0.localSizeLong(part, OFFHEAP, BACKUP)); |
| |
| int globalParitionSwapPrimary = 0; |
| int globalPartSwapBackup = 0; |
| |
| int globalPartOffheapPrimary = 0; |
| int globalPartOffheapBackup = 0; |
| |
| for (int i = 0; i < gridCount(); i++) { |
| T2<Integer, Integer> offheap = offheapKeysCount(i, part); |
| |
| globalPartOffheapPrimary += offheap.get1(); |
| globalPartOffheapBackup += offheap.get2(); |
| } |
| |
| int backups; |
| |
| if (cacheMode() == PARTITIONED) |
| backups = 1; |
| else // REPLICATED. |
| backups = gridCount() - 1; |
| |
| int globalTotal = totalKeys + totalKeys * backups; |
| int globalPartTotalSwap = globalParitionSwapPrimary + globalPartSwapBackup; |
| int globalPartTotalOffheap = globalPartOffheapPrimary + globalPartOffheapBackup; |
| |
| log.info("Global keys [total=" + globalTotal + |
| ", offheap=" + globalPartTotalOffheap + |
| ", swap=" + globalPartTotalSwap + ']'); |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(totalKeys, cache.size(PRIMARY)); |
| assertEquals(globalTotal, cache.size(ALL)); |
| assertEquals(globalTotal, cache.size(PRIMARY, BACKUP, NEAR, ONHEAP, OFFHEAP)); |
| assertEquals(globalTotal, cache.size(ONHEAP, OFFHEAP, PRIMARY, BACKUP)); |
| |
| assertEquals(globalPartTotalSwap, cache.sizeLong(part, PRIMARY, BACKUP, NEAR)); |
| assertEquals(globalParitionSwapPrimary, cache.sizeLong(part, PRIMARY)); |
| assertEquals(globalPartSwapBackup, cache.sizeLong(part, BACKUP)); |
| |
| assertEquals(globalPartTotalOffheap, cache.sizeLong(part, PRIMARY, BACKUP, NEAR, OFFHEAP)); |
| assertEquals(globalPartOffheapPrimary, cache.sizeLong(part, OFFHEAP, PRIMARY)); |
| assertEquals(globalPartOffheapBackup, cache.sizeLong(part, OFFHEAP, BACKUP)); |
| |
| assertEquals(globalPartTotalSwap + globalPartTotalOffheap, cache.sizeLong(part, PRIMARY, BACKUP, NEAR, OFFHEAP)); |
| assertEquals(globalParitionSwapPrimary + globalPartOffheapPrimary, cache.sizeLong(part, OFFHEAP, PRIMARY)); |
| assertEquals(globalPartSwapBackup + globalPartOffheapBackup, cache.sizeLong(part, OFFHEAP, BACKUP)); |
| } |
| } |
| finally { |
| cache0.removeAll(new HashSet<>(primaryKeys)); |
| cache0.removeAll(new HashSet<>(backupKeys)); |
| } |
| |
| checkEmpty(); |
| } |
| |
| /** |
| * @param exp Expected size. |
| */ |
| private void checkPrimarySize(int exp) { |
| int size = 0; |
| |
| for (int i = 0; i < gridCount(); i++) { |
| IgniteCache<Integer, String> cache = jcache(i); |
| |
| assertEquals(exp, cache.size(PRIMARY)); |
| |
| size += cache.localSize(PRIMARY); |
| |
| assertEquals(exp, (int)cache.sizeAsync(PRIMARY).get()); |
| } |
| |
| assertEquals(exp, size); |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkLocalEntriesStorageFilter(int nodeIdx) throws Exception { |
| if (true) // TODO GG-11148. |
| return; |
| |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| List<Integer> primaryKeys = primaryKeys(cache0, 100, 10_000); |
| List<Integer> backupKeys = backupKeys(cache0, 100, 10_000); |
| |
| try { |
| final String val = "test_value-"; |
| |
| for (int i = 0; i < 100; i++) { |
| cache0.put(primaryKeys.get(i), val + primaryKeys.get(i)); |
| cache0.put(backupKeys.get(i), val + backupKeys.get(i)); |
| } |
| |
| int totalKeys = 200; |
| |
| T2<List<Integer>, List<Integer>> swapKeys = swapKeys(nodeIdx); |
| |
| assertTrue(!swapKeys.get1().isEmpty()); |
| assertTrue(!swapKeys.get2().isEmpty()); |
| |
| T2<List<Integer>, List<Integer>> offheapKeys = offheapKeys(nodeIdx); |
| |
| assertTrue(!offheapKeys.get1().isEmpty()); |
| assertTrue(!offheapKeys.get2().isEmpty()); |
| |
| List<Integer> swap = new ArrayList<>(); |
| |
| swap.addAll(swapKeys.get1()); |
| swap.addAll(swapKeys.get2()); |
| |
| assertFalse(swap.isEmpty()); |
| |
| List<Integer> offheap = new ArrayList<>(); |
| |
| offheap.addAll(offheapKeys.get1()); |
| offheap.addAll(offheapKeys.get2()); |
| |
| assertFalse(offheap.isEmpty()); |
| |
| List<Integer> heap = new ArrayList<>(); |
| |
| heap.addAll(primaryKeys); |
| heap.addAll(backupKeys); |
| |
| heap.removeAll(swap); |
| heap.removeAll(offheap); |
| |
| log.info("Keys [total=" + totalKeys + |
| ", offheap=" + offheap.size() + |
| ", swap=" + swap.size() + ']'); |
| |
| assertFalse(heap.isEmpty()); |
| |
| checkLocalEntries(cache0.localEntries(), val, primaryKeys, backupKeys); |
| checkLocalEntries(cache0.localEntries(ALL), val, primaryKeys, backupKeys); |
| checkLocalEntries(cache0.localEntries(ONHEAP, OFFHEAP), val, primaryKeys, backupKeys); |
| |
| checkLocalEntries(cache0.localEntries(), val, swap); |
| checkLocalEntries(cache0.localEntries(OFFHEAP), val, offheap); |
| checkLocalEntries(cache0.localEntries(ONHEAP), val, heap); |
| |
| checkLocalEntries(cache0.localEntries(OFFHEAP), val, swap, offheap); |
| checkLocalEntries(cache0.localEntries(ONHEAP), val, swap, heap); |
| |
| checkLocalEntries(cache0.localEntries(PRIMARY), val, swapKeys.get1()); |
| checkLocalEntries(cache0.localEntries(BACKUP), val, swapKeys.get2()); |
| checkLocalEntries(cache0.localEntries(OFFHEAP, PRIMARY), val, offheapKeys.get1()); |
| checkLocalEntries(cache0.localEntries(OFFHEAP, BACKUP), val, offheapKeys.get2()); |
| |
| checkLocalEntries(cache0.localEntries(OFFHEAP, PRIMARY), val, swapKeys.get1(), offheapKeys.get1()); |
| checkLocalEntries(cache0.localEntries(OFFHEAP, BACKUP), val, swapKeys.get2(), offheapKeys.get2()); |
| checkLocalEntries(cache0.localEntries(OFFHEAP, PRIMARY, BACKUP), val, swap, offheap); |
| } |
| finally { |
| cache0.removeAll(new HashSet<>(primaryKeys)); |
| cache0.removeAll(new HashSet<>(backupKeys)); |
| } |
| } |
| |
| /** |
| * @param nodeIdx Node index. |
| * @throws Exception If failed. |
| */ |
| private void checkLocalEntriesAffinityFilter(int nodeIdx) throws Exception { |
| IgniteCache<Integer, String> cache0 = jcache(nodeIdx); |
| |
| final int PUT_KEYS = 10; |
| |
| List<Integer> primaryKeys = null; |
| List<Integer> backupKeys = null; |
| List<Integer> nearKeys = null; |
| |
| try { |
| primaryKeys = primaryKeys(cache0, PUT_KEYS, 0); |
| backupKeys = backupKeys(cache0, PUT_KEYS, 0); |
| |
| for (Integer key : primaryKeys) |
| cache0.put(key, String.valueOf(key)); |
| for (Integer key : backupKeys) |
| cache0.put(key, String.valueOf(key)); |
| |
| nearKeys = cacheMode() == PARTITIONED ? nearKeys(cache0, PUT_KEYS, 0) : Collections.<Integer>emptyList(); |
| |
| for (Integer key : nearKeys) |
| cache0.put(key, String.valueOf(key)); |
| |
| log.info("Keys [near=" + nearKeys + ", primary=" + primaryKeys + ", backup=" + backupKeys + ']'); |
| |
| if (hasNearCache()) { |
| checkLocalEntries(cache0.localEntries(), nearKeys, primaryKeys, backupKeys); |
| checkLocalEntries(cache0.localEntries(ALL), nearKeys, primaryKeys, backupKeys); |
| checkLocalEntries(cache0.localEntries(NEAR), nearKeys); |
| checkLocalEntries(cache0.localEntries(PRIMARY, BACKUP, NEAR), nearKeys, primaryKeys, backupKeys); |
| checkLocalEntries(cache0.localEntries(NEAR, PRIMARY), nearKeys, primaryKeys); |
| checkLocalEntries(cache0.localEntries(NEAR, BACKUP), nearKeys, backupKeys); |
| } |
| else { |
| checkLocalEntries(cache0.localEntries(), primaryKeys, backupKeys); |
| checkLocalEntries(cache0.localEntries(ALL), primaryKeys, backupKeys); |
| checkLocalEntries(cache0.localEntries(NEAR)); |
| checkLocalEntries(cache0.localEntries(NEAR, PRIMARY), primaryKeys); |
| checkLocalEntries(cache0.localEntries(NEAR, BACKUP), backupKeys); |
| checkLocalEntries(cache0.localEntries(PRIMARY, BACKUP, NEAR), primaryKeys, backupKeys); |
| } |
| |
| checkLocalEntries(cache0.localEntries(PRIMARY), primaryKeys); |
| checkLocalEntries(cache0.localEntries(BACKUP), backupKeys); |
| checkLocalEntries(cache0.localEntries(PRIMARY, BACKUP), primaryKeys, backupKeys); |
| } |
| finally { |
| if (primaryKeys != null) |
| cache0.removeAll(new HashSet<>(primaryKeys)); |
| |
| if (backupKeys != null) |
| cache0.removeAll(new HashSet<>(backupKeys)); |
| |
| if (nearKeys != null) |
| cache0.removeAll(new HashSet<>(nearKeys)); |
| } |
| } |
| |
| /** |
| * @param entries Entries. |
| * @param exp Expected entries. |
| */ |
| private void checkLocalEntries(Iterable<Cache.Entry<Integer, String>> entries, Collection<Integer>... exp) { |
| checkLocalEntries(entries, "", exp); |
| } |
| |
| /** |
| * @param entries Entries. |
| * @param expVal Expected value. |
| * @param exp Expected keys. |
| */ |
| private void checkLocalEntries(Iterable<Cache.Entry<Integer, String>> entries, |
| String expVal, |
| Collection<Integer>... exp) { |
| Set<Integer> allExp = new HashSet<>(); |
| |
| for (Collection<Integer> col : exp) |
| assertTrue(allExp.addAll(col)); |
| |
| for (Cache.Entry<Integer, String> e : entries) { |
| assertNotNull(e.getKey()); |
| assertNotNull(e.getValue()); |
| assertEquals(expVal + e.getKey(), e.getValue()); |
| |
| assertTrue("Unexpected entry: " + e, allExp.remove(e.getKey())); |
| } |
| |
| assertTrue("Expected entries not found: " + allExp, allExp.isEmpty()); |
| } |
| } |
