/**
 * Refreshes table stats if they are outdated.
 */
private void refreshStatsIfNeeded() {
    TableStatistics stats = tblStats;

    long statsTotalRowCnt = stats.totalRowCount();
    long curTotalRowCnt = size.sum();

    // Update stats if total table size changed significantly since the last stats update.
    if (needRefreshStats(statsTotalRowCnt, curTotalRowCnt) && cacheInfo.affinityNode()) {
        long primaryRowCnt = cacheSize(CachePeekMode.PRIMARY);
        long totalRowCnt = cacheSize(CachePeekMode.PRIMARY, CachePeekMode.BACKUP);

        size.reset();
        size.add(totalRowCnt);

        tblStats = new TableStatistics(totalRowCnt, primaryRowCnt);
    }
}
 

/**
 * @throws Exception If test failed.
 */
@Test
public void testGlobalClearAll() throws Exception {
    cache1.put(1, "val1");
    cache2.put(2, "val2");
    cache3.put(3, "val3");

    assertEquals(3, cache1.localSize(CachePeekMode.ALL));
    assertEquals(3, cache2.localSize(CachePeekMode.ALL));
    assertEquals(3, cache3.localSize(CachePeekMode.ALL));

    cache1.clear();

    assertEquals(0, cache1.localSize(CachePeekMode.ALL));
    assertEquals(0, cache2.localSize(CachePeekMode.ALL));
    assertEquals(0, cache3.localSize(CachePeekMode.ALL));
}
 

/**
 * @param c Cache.
 * @param cnt Key count.
 */
private void checkKeys(IgniteCache<Integer, String> c, int cnt) {
    Affinity<Integer> aff = affinity(c);

    boolean sync = isSync(c);

    Ignite ignite = c.unwrap(Ignite.class);

    for (int i = 0; i < cnt; i++) {
        if (aff.mapPartitionToPrimaryAndBackups(aff.partition(i)).contains(ignite.cluster().localNode())) {
            String val = sync ? c.localPeek(i, CachePeekMode.ONHEAP) : c.get(i);

            assertEquals("Key check failed [igniteInstanceName=" + ignite.name() + ", cache=" + c.getName() +
                    ", key=" + i + ']', Integer.toString(i), val);
        }
    }
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testLoadCache() throws Exception {
    IgniteCache<Integer, Integer> cache = jcache(0);

    final Integer key = primaryKey(cache);

    storeMap.put(key, 100);

    try {
        cache.loadCache(null);

        checkTtl(key, 500, false);

        assertEquals((Integer)100, cache.localPeek(key, CachePeekMode.ONHEAP));

        U.sleep(600);

        checkExpired(key);
    }
    finally {
        cache.removeAll();
    }
}
 

/** @throws Exception If failed. */
@Test
public void testNoTransactionWriteThrough() throws Exception {
    IgniteCache<Integer, String> near = jcache(0);

    near.put(2, "2");

    String s = near.getAndPut(3, "3");

    assertNotNull(s);
    assertEquals("3", s);

    assertEquals("2", near.localPeek(2));
    assertEquals("3", near.localPeek(3));

    assertEquals("2", near.get(2));
    assertEquals("3", near.get(3));

    assertEquals("2", localPeek(dht(primaryGrid(2)), 2));
    assertEquals("3", localPeek(dht(primaryGrid(3)), 3));

    assertEquals(2, near.localSize(CachePeekMode.ALL));
    assertEquals(2, near.localSize(CachePeekMode.ALL));
}
 

/**
 * @param atomicityMode Atomicity mode.
 * @param cacheMode Cache mode.
 * @throws Exception If failed.
 */
private void testEvictionWithReadThrough(CacheAtomicityMode atomicityMode, CacheMode cacheMode) throws Exception {
    startGrid(0);

    CacheConfiguration<Object, Object> cfg = cacheConfig("evict-rebalance", null, cacheMode, atomicityMode,
        CacheWriteSynchronizationMode.PRIMARY_SYNC);
    cfg.setReadThrough(true);
    cfg.setCacheStoreFactory(new TestStoreFactory());

    IgniteCache<Object, Object> cache = ignite(0).getOrCreateCache(cfg);

    for (int i = 1; i <= ENTRIES; i++) {
        cache.get(i);

        if (i % (ENTRIES / 10) == 0)
            System.out.println(">>> Entries: " + i);
    }

    int size = cache.size(CachePeekMode.PRIMARY);

    System.out.println(">>> Resulting size: " + size);

    assertTrue(size > 0);

    assertTrue(size < ENTRIES);
}
 

/** */
private void printSizesDataNodes(int nodesCnt, int emptyNodeIdx) {
    for (int i = 0; i < nodesCnt; i++) {
        IgniteEx ig = grid(i);

        int locSize = ig.cache(CACHE_NAME).localSize(CachePeekMode.PRIMARY);

        if (i == emptyNodeIdx)
            assertEquals("Cache local size on "
                + i
                + " node is expected to be zero", 0, locSize);
        else
            assertTrue("Cache local size on "
                + i
                + " node is expected to be non zero", locSize > 0);
    }
}
 

/**
 * @param keys Keys.
 * @throws Exception If failed.
 */
public void checkCache(Object[] keys) throws Exception {
    preloadMode = SYNC;

    lifecycleBean = lifecycleBean(keys);

    for (int i = 0; i < gridCnt; i++) {
        startGrid(i);

        info("Checking '" + (i + 1) + "' nodes...");

        for (int j = 0; j < G.allGrids().size(); j++) {
            IgniteCache<String, MyValue> c1 = grid(j).cache("one");
            IgniteCache<String, MyValue> c2 = grid(j).cache("two");

            int size1 = c1.localSize(CachePeekMode.ALL);
            int size2 = c2.localSize(CachePeekMode.ALL);

            assertEquals(" Invalid cache1 size [i=" + i + ", j=" + j + ", size=" + size1 + ']', keys.length, size1);
            assertEquals(" Invalid cache2 size [i=" + i + ", j=" + j + ", size=" + size2 + ']', keys.length / 2, size2);
        }
    }
}
 

/** {@inheritDoc} */
@Override public int size(CachePeekMode... peekModes) throws CacheException {
    IgniteInternalCache<K, V> delegate = getDelegateSafe();

    try {
        if (isAsync()) {
            setFuture(delegate.sizeAsync(peekModes));

            return 0;
        }
        else
            return delegate.size(peekModes);
    }
    catch (IgniteCheckedException | IgniteException e) {
        throw cacheException(e);
    }
}
 

/** {@inheritDoc} */
@Override protected void afterTest() throws Exception {
    TestIndexingSpi.forceFail(false);

    Transaction tx = jcache().unwrap(Ignite.class).transactions().tx();

    if (tx != null) {
        tx.close();

        fail("Cache transaction remained after test completion: " + tx);
    }

    for (int key = 0; key <= lastKey; key++)
        grid(0).cache(DEFAULT_CACHE_NAME).remove(key);

    assertEquals(0, jcache(0).size(CachePeekMode.ALL));
}
 

/**
 * @throws Exception If test failed.
 */
@Test
public void testSync() throws Exception {
    preloadMode = SYNC;
    batchSize = 512;

    try {
        IgniteCache<Integer, String> cache1 = startGrid(1).cache(DEFAULT_CACHE_NAME);

        int keyCnt = 1000;

        for (int i = 0; i < keyCnt; i++)
            cache1.put(i, "val" + i);

        IgniteCache<Integer, String> cache2 = startGrid(2).cache(DEFAULT_CACHE_NAME);

        assertEquals(keyCnt, cache2.localSize(CachePeekMode.ALL));
    }
    finally {
        stopAllGrids();
    }
}
 

/**
 * @param idx Node index.
 * @param cacheName Cache name.
 * @param data Expected data.
 * @throws Exception If failed.
 */
private void checkLocalData(int idx, String cacheName, Integer[] data) throws Exception {
    Ignite ignite = ignite(idx);
    ClusterNode node = ignite.cluster().localNode();
    IgniteCache cache = ignite.<Integer, Integer>cache(cacheName);

    Affinity aff = affinity(cache);

    Set<Integer> locKeys = new TreeSet<>();

    for (int key = 0; key < data.length; key++) {
        if (aff.isPrimaryOrBackup(node, key))
            locKeys.add(key);
    }

    Iterable<Cache.Entry<Integer, Integer>> locEntries = cache.localEntries(CachePeekMode.OFFHEAP);

    for (Cache.Entry<Integer, Integer> entry : locEntries) {
        assertTrue(locKeys.remove(entry.getKey()));
        assertEquals(data[entry.getKey()], entry.getValue());
    }

    assertTrue(locKeys.isEmpty());
}
 

@Override
public long getNumberOfEntities(SessionFactory sessionFactory) {
	int entityCount = 0;
	Set<IgniteCache<?, ?>> processedCaches = Collections.newSetFromMap( new IdentityHashMap<IgniteCache<?, ?>, Boolean>() );
	for ( EntityPersister entityPersister : ( (SessionFactoryImplementor) sessionFactory ).getEntityPersisters().values() ) {
		IgniteCache<?, ?> entityCache = getEntityCache( sessionFactory, ( (OgmEntityPersister) entityPersister ).getEntityKeyMetadata() );
		if ( !processedCaches.contains( entityCache ) ) {
			entityCount += entityCache.size( CachePeekMode.ALL );
			processedCaches.add( entityCache );
		}
	}
	return entityCount;
}
 

/**
 *  Tries to propagate cache with binary objects created using the builder.
 *
 * @throws Exception If failed.
 */
@Test
public void testAddBinaryCreatedWithBuilder() throws Exception {
    try {
        binaries = true;

        startGrids(2);

        awaitPartitionMapExchange();

        Ignite g0 = grid(0);

        IgniteDataStreamer<Integer, BinaryObject> dataLdr = g0.dataStreamer(DEFAULT_CACHE_NAME);

        for (int i = 0; i < 500; i++) {
            BinaryObjectBuilder obj = g0.binary().builder("NoExistedClass");

            obj.setField("id", i);
            obj.setField("name", "name = " + i);

            dataLdr.addData(i, obj.build());
        }

        dataLdr.close(false);

        assertEquals(500, g0.cache(DEFAULT_CACHE_NAME).size(CachePeekMode.ALL));
        assertEquals(500, grid(1).cache(DEFAULT_CACHE_NAME).size(CachePeekMode.ALL));
    }
    finally {
        G.stopAll(true);
    }
}
 

/**
 * @param cfg Config.
 * @throws Exception If failed.
 */
protected void createCacheAndTestEviction(CacheConfiguration<Object, Object> cfg) throws Exception {
    IgniteCache<Object, Object> cache = clientGrid().getOrCreateCache(cfg);

    for (int i = 1; i <= ENTRIES; i++) {
        ThreadLocalRandom r = ThreadLocalRandom.current();

        if (r.nextInt() % 5 == 0)
            cache.put(i, new TestObject(PAGE_SIZE / 4 - 50 + r.nextInt(5000))); // Fragmented object.
        else
            cache.put(i, new TestObject(r.nextInt(PAGE_SIZE / 4 - 50))); // Fits in one page.

        if (r.nextInt() % 7 == 0)
            cache.get(r.nextInt(i)); // Touch.
        else if (r.nextInt() % 11 == 0)
            cache.remove(r.nextInt(i)); // Remove.
        else if (r.nextInt() % 13 == 0)
            cache.put(r.nextInt(i), new TestObject(r.nextInt(PAGE_SIZE / 2))); // Update.

        if (i % (ENTRIES / 10) == 0)
            System.out.println(">>> Entries put: " + i);
    }

    int resultingSize = cache.size(CachePeekMode.PRIMARY);

    System.out.println(">>> Resulting size: " + resultingSize);

    // Eviction started, no OutOfMemory occurred, success.
    assertTrue(resultingSize < ENTRIES * 10 / 11);

    clientGrid().destroyCache(cfg.getName());
}
 

/**
 * Retrieves partitions size.
 *
 * @return Rows count.
 */
private long cacheSize(CachePeekMode... modes) {
    try {
        return cacheInfo.cacheContext().cache().localSize(modes);
    }
    catch (IgniteCheckedException e) {
        throw new IgniteException(e);
    }
}
 

/** @throws Exception If failed. */
@Test
public void testConcurrentOps() throws Exception {
    // Don't create missing values.
    store.create(false);

    IgniteCache<Integer, String> near = jcache(0);

    int key = 1;

    assertTrue(near.putIfAbsent(key, "1"));
    assertFalse(near.putIfAbsent(key, "1"));
    assertEquals("1", near.getAndPutIfAbsent(key, "2"));

    assertEquals("1", near.localPeek(key));
    assertEquals(1, near.localSize(CachePeekMode.ALL));
    assertEquals(1, near.localSize(CachePeekMode.ALL));

    assertEquals("1", near.getAndReplace(key, "2"));
    assertEquals("2", near.localPeek(key));

    assertTrue(near.replace(key, "2"));

    assertEquals("2", near.localPeek(key));
    assertEquals(1, near.localSize(CachePeekMode.ALL));
    assertEquals(1, near.localSize(CachePeekMode.ALL));

    assertTrue(near.remove(key, "2"));

    assertEquals(0, near.localSize(CachePeekMode.ALL));
}
 

/**
 * Execute given query locally and check results.
 * @param cache Cache to run query on.
 * @param fldsQry Query.
 */
private void doTestLocalQuery(IgniteCache<?, ?> cache, SqlFieldsQuery fldsQry) throws InterruptedException {
    awaitPartitionMapExchange(true, true, null);

    int exp = 0;

    for (Cache.Entry e: intCache.localEntries(CachePeekMode.PRIMARY)) {
        if (e.getValue() instanceof Integer)
            exp++;
    }

    QueryCursor<List<?>> qry = cache.query(fldsQry.setLocal(true));

    assertEquals(exp, qry.getAll().size());
}
 

/** {@inheritDoc} */
@Nullable @Override public Object localPeek(
    Object key,
    CachePeekMode[] peekModes
) throws IgniteCheckedException {
    return delegate.get().localPeek(keyTransformer.transform(key), peekModes);
}
 

/**
 * Checks if keys are present.
 *
 * @param c Cache.
 * @param keyCnt Key count.
 */
private void checkCache(IgniteCache<String, Integer> c, int keyCnt) {
    Ignite g = c.unwrap(Ignite.class);

    for (int i = 0; i < keyCnt; i++) {
        String key = Integer.toString(i);

        if (affinity(c).isPrimaryOrBackup(g.cluster().localNode(), key))
            assertEquals(Integer.valueOf(i), c.localPeek(key, CachePeekMode.ONHEAP));
    }
}
 

/**
 * @param cnt Count.
 */
private void assertCacheEntriesLoaded(int cnt) {
    // get the cache and check that the entries are present
    IgniteCache<Integer, String> cache = grid().cache(DEFAULT_CACHE_NAME);

    // for each key from 0 to count from the TEST_DATA (ordered by key), check that the entry is present in cache
    for (Integer key : new ArrayList<>(new TreeSet<>(TEST_DATA.keySet())).subList(0, cnt))
        assertEquals(TEST_DATA.get(key), cache.get(key));

    // assert that the cache exactly the specified amount of elements
    assertEquals(cnt, cache.size(CachePeekMode.ALL));

    // remove the event listener
    grid().events(grid().cluster().forCacheNodes(DEFAULT_CACHE_NAME)).stopRemoteListen(remoteLsnr);
}
 

/** @throws Exception If failed. */
@Test
public void testSingleLock() throws Exception {
    IgniteCache<Integer, String> near = jcache();

    Lock lock = near.lock(1);

    lock.lock();

    try {
        near.put(1, "1");

        assertEquals("1", near.localPeek(1, CachePeekMode.ONHEAP));
        assertEquals("1", dhtPeek(1));

        assertEquals("1", near.get(1));
        assertEquals("1", near.getAndRemove(1));

        assertNull(near.localPeek(1, CachePeekMode.ONHEAP));
        assertNull(dhtPeek(1));

        assertTrue(near.isLocalLocked(1, false));
        assertTrue(near.isLocalLocked(1, true));
    }
    finally {
        lock.unlock();
    }

    assertFalse(near.isLocalLocked(1, false));
    assertFalse(near.isLocalLocked(1, true));
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testEvictAll() throws Exception {
    FifoEvictionPolicy<Object, Object> plc = new FifoEvictionPolicy<>();
    plc.setMaxSize(500);

    this.plc = plc;

    try {
        Ignite ignite = startGrid(1);

        final IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME);

        Collection<Integer> keys = new ArrayList<>(100);

        for (int i = 0; i < 100; i++) {
            cache.put(i, i);

            keys.add(i);
        }

        assertEquals(100, ((FifoEvictionPolicy)plc).queue().size());

        for (Integer key : keys)
            cache.localEvict(Collections.singleton(key));

        assertEquals(0, ((FifoEvictionPolicy)plc).queue().size());
        assertEquals(0, cache.size(CachePeekMode.ONHEAP));
    }
    finally {
        stopAllGrids();
    }
}
 

/**
 * Removes old cache entries locally.
 *
 * @param ignite Ignite.
 * @param dataCacheName Cache name.
 * @param currentVersions Current versions.
 */
private static void removeOldEntries(final Ignite ignite, final String dataCacheName,
    final Map<String, EntryProcessorResult<Long>> currentVersions) {

    IgniteCache<PlatformDotNetEntityFrameworkCacheKey, PlatformDotNetEntityFrameworkCacheEntry> cache =
        ignite.cache(dataCacheName);

    Set<PlatformDotNetEntityFrameworkCacheKey> keysToRemove = new TreeSet<>();

    ClusterNode localNode = ignite.cluster().localNode();

    for (Cache.Entry<PlatformDotNetEntityFrameworkCacheKey, PlatformDotNetEntityFrameworkCacheEntry> cacheEntry :
        cache.localEntries(CachePeekMode.ALL)) {
        // Check if we are on a primary node for the key, since we use CachePeekMode.ALL
        // and we don't want to process backup entries.
        if (!ignite.affinity(dataCacheName).isPrimary(localNode, cacheEntry.getKey()))
            continue;

        long[] versions = cacheEntry.getKey().versions();
        String[] entitySets = cacheEntry.getValue().entitySets();

        for (int i = 0; i < entitySets.length; i++) {
            EntryProcessorResult<Long> curVer = currentVersions.get(entitySets[i]);

            if (curVer != null && versions[i] < curVer.get())
                keysToRemove.add(cacheEntry.getKey());
        }
    }

    cache.removeAll(keysToRemove);
}
 

/** {@inheritDoc} */
@Override public void run() {
    IgniteLogger log = localNode.log();

    log.info("Start local rebalancing caches");

    for (String cacheName : localNode.cacheNames()) {
        IgniteCache<?, ?> cache = localNode.cache(cacheName);

        assertNotNull(cache);

        boolean finished;

        log.info("Start rebalancing cache: " + cacheName + ", size: " + cache.localSize());

        do {
            IgniteFuture<?> rebalance = cache.rebalance();

            log.info("Wait rebalancing cache: " + cacheName + " - " + rebalance);

            finished = (Boolean)rebalance.get();

            log.info("Rebalancing cache: " + cacheName + " - " + rebalance);

            if (finished) {
                log.info("Finished rebalancing cache: " + cacheName + ", size: " +
                    cache.localSize(CachePeekMode.PRIMARY) + cache.localSize(CachePeekMode.BACKUP));
            } else
                log.info("Rescheduled rebalancing cache: " + cacheName + ", size: " + cache.localSize());
        }
        while (!finished);
    }

    log.info("Finished local rebalancing caches");
}
 

/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override public int size() {
    try {
        onAccess();

        if (separated) {
            // Non collocated IgniteSet uses a separate cache which contains additional header element.
            return cache.sizeAsync(new CachePeekMode[] {}).get() - 1;
        }

        CacheQuery qry = new GridCacheQueryAdapter<>(ctx, SET, null, null,
            new GridSetQueryPredicate<>(id, collocated), collocated ? hdrPart : null,
            false, false, null);

        Collection<ClusterNode> nodes = dataNodes(ctx.affinity().affinityTopologyVersion());

        qry.projection(ctx.grid().cluster().forNodes(nodes));

        CacheQueryFuture<Integer> qryFut = qry.execute(new SumReducer());

        int sum = 0;

        Integer val;

        while ((val = qryFut.next()) != null)
            sum += val;

        return sum;
    }
    catch (IgniteCheckedException e) {
        throw U.convertException(e);
    }
}
 

/** {@inheritDoc} */
@Override public Iterable<Entry<K, V>> localEntries(CachePeekMode... peekModes) throws CacheException {
    CacheOperationGate opGate = onEnter();

    try {
        return delegate.localEntries(peekModes);
    }
    finally {
        onLeave(opGate);
    }
}
 

/** {@inheritDoc} */
@Override public V localPeek(K key, CachePeekMode... peekModes) {
    CacheOperationGate opGate = onEnter();

    try {
        return delegate.localPeek(key, peekModes);
    }
    finally {
        onLeave(opGate);
    }
}
 

/** {@inheritDoc} */
@Override public long sizeLong(CachePeekMode... peekModes) throws CacheException {
    CacheOperationGate opGate = onEnter();

    try {
        return delegate.sizeLong(peekModes);
    }
    finally {
        onLeave(opGate);
    }
}
 

/** {@inheritDoc} */
@Override public long sizeLong(int partition, CachePeekMode... peekModes) throws CacheException {
    CacheOperationGate opGate = onEnter();

    try {
        return delegate.sizeLong(partition, peekModes);
    }
    finally {
        onLeave(opGate);
    }
}