ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yzhda...@apache.org
Subject [38/50] [abbrv] incubator-ignite git commit: Merge branches 'ignite-394' and 'sprint-2' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-394
Date Wed, 11 Mar 2015 14:54:25 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/00fd3c3c/modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerPerformanceTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerPerformanceTest.java
index 33fe310,0000000..c27e7c6
mode 100644,000000..100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerPerformanceTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerPerformanceTest.java
@@@ -1,199 -1,0 +1,197 @@@
 +/*
 + * 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.datastream;
 +
 +import org.apache.ignite.*;
 +import org.apache.ignite.configuration.*;
 +import org.apache.ignite.internal.util.typedef.internal.*;
 +import org.apache.ignite.spi.discovery.tcp.*;
 +import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
 +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
 +import org.apache.ignite.testframework.junits.common.*;
 +import org.jdk8.backport.*;
 +
 +import java.util.concurrent.*;
 +
 +import static org.apache.ignite.cache.CacheDistributionMode.*;
 +import static org.apache.ignite.cache.CacheMode.*;
 +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
 +import static org.apache.ignite.events.EventType.*;
 +
 +/**
 + * Data streamer performance test. Compares group lock data streamer to traditional lock.
 + * <p>
 + * Disable assertions and give at least 2 GB heap to run this test.
 + */
 +public class IgniteDataStreamerPerformanceTest extends GridCommonAbstractTest {
 +    /** */
 +    private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
 +
 +    /** */
 +    private static final int GRID_CNT = 3;
 +
 +    /** */
 +    private static final int ENTRY_CNT = 80000;
 +
 +    /** */
 +    private boolean useCache;
 +
 +    /** */
 +    private String[] vals = new String[2048];
 +
 +    /** {@inheritDoc} */
 +    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
 +        IgniteConfiguration cfg = super.getConfiguration(gridName);
 +
 +        TcpDiscoverySpi spi = new TcpDiscoverySpi();
 +
 +        spi.setIpFinder(IP_FINDER);
 +
 +        cfg.setDiscoverySpi(spi);
 +
 +        cfg.setIncludeProperties();
 +
 +        cfg.setIncludeEventTypes(EVT_TASK_FAILED, EVT_TASK_FINISHED, EVT_JOB_MAPPED);
 +
 +        cfg.setConnectorConfiguration(null);
 +
 +        cfg.setPeerClassLoadingEnabled(true);
 +
 +        if (useCache) {
 +            CacheConfiguration cc = defaultCacheConfiguration();
 +
 +            cc.setCacheMode(PARTITIONED);
 +
 +            cc.setDistributionMode(PARTITIONED_ONLY);
 +            cc.setWriteSynchronizationMode(FULL_SYNC);
 +            cc.setStartSize(ENTRY_CNT / GRID_CNT);
 +            cc.setSwapEnabled(false);
 +
 +            cc.setBackups(1);
 +
-             cc.setStoreValueBytes(true);
- 
 +            cfg.setCacheSanityCheckEnabled(false);
 +            cfg.setCacheConfiguration(cc);
 +        }
 +        else
 +            cfg.setCacheConfiguration();
 +
 +        return cfg;
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void beforeTestsStarted() throws Exception {
 +        super.beforeTestsStarted();
 +
 +        for (int i = 0; i < vals.length; i++) {
 +            int valLen = ThreadLocalRandom8.current().nextInt(128, 512);
 +
 +            StringBuilder sb = new StringBuilder();
 +
 +            for (int j = 0; j < valLen; j++)
 +                sb.append('a' + ThreadLocalRandom8.current().nextInt(20));
 +
 +            vals[i] = sb.toString();
 +
 +            info("Value: " + vals[i]);
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testPerformance() throws Exception {
 +        doTest();
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    private void doTest() throws Exception {
 +        System.gc();
 +        System.gc();
 +        System.gc();
 +
 +        try {
 +            useCache = true;
 +
 +            startGridsMultiThreaded(GRID_CNT);
 +
 +            useCache = false;
 +
 +            Ignite ignite = startGrid();
 +
 +            final IgniteDataStreamer<Integer, String> ldr = ignite.dataStreamer(null);
 +
 +            ldr.perNodeBufferSize(8192);
 +            ldr.updater(IgniteDataStreamerCacheUpdaters.<Integer, String>batchedSorted());
 +            ldr.autoFlushFrequency(0);
 +
 +            final LongAdder cnt = new LongAdder();
 +
 +            long start = U.currentTimeMillis();
 +
 +            Thread t = new Thread(new Runnable() {
 +                @SuppressWarnings("BusyWait")
 +                @Override public void run() {
 +                    while (true) {
 +                        try {
 +                            Thread.sleep(10000);
 +                        }
 +                        catch (InterruptedException ignored) {
 +                            break;
 +                        }
 +
 +                        info(">>> Adds/sec: " + cnt.sumThenReset() / 10);
 +                    }
 +                }
 +            });
 +
 +            t.setDaemon(true);
 +
 +            t.start();
 +
 +            int threadNum = 2;//Runtime.getRuntime().availableProcessors();
 +
 +            multithreaded(new Callable<Object>() {
 +                @SuppressWarnings("InfiniteLoopStatement")
 +                @Override public Object call() throws Exception {
 +                    ThreadLocalRandom8 rnd = ThreadLocalRandom8.current();
 +
 +                    while (true) {
 +                        int i = rnd.nextInt(ENTRY_CNT);
 +
 +                        ldr.addData(i, vals[rnd.nextInt(vals.length)]);
 +
 +                        cnt.increment();
 +                    }
 +                }
 +            }, threadNum, "loader");
 +
 +            info("Closing loader...");
 +
 +            ldr.close(false);
 +
 +            long duration = U.currentTimeMillis() - start;
 +
 +            info("Finished performance test. Duration: " + duration + "ms.");
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/00fd3c3c/modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerProcessorSelfTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerProcessorSelfTest.java
index 9402c0c,0000000..2996426
mode 100644,000000..100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerProcessorSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerProcessorSelfTest.java
@@@ -1,924 -1,0 +1,970 @@@
 +/*
 + * 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.datastream;
 +
 +import org.apache.ignite.*;
 +import org.apache.ignite.cache.*;
 +import org.apache.ignite.cache.affinity.*;
 +import org.apache.ignite.cache.store.*;
 +import org.apache.ignite.cluster.*;
 +import org.apache.ignite.configuration.*;
 +import org.apache.ignite.events.*;
 +import org.apache.ignite.internal.*;
 +import org.apache.ignite.internal.processors.cache.*;
 +import org.apache.ignite.internal.processors.cache.distributed.near.*;
 +import org.apache.ignite.internal.util.typedef.internal.*;
 +import org.apache.ignite.lang.*;
 +import org.apache.ignite.marshaller.optimized.*;
 +import org.apache.ignite.spi.discovery.tcp.*;
 +import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
 +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
 +import org.apache.ignite.testframework.junits.common.*;
 +import org.jetbrains.annotations.*;
 +
 +import javax.cache.*;
 +import javax.cache.configuration.*;
 +import java.util.*;
 +import java.util.concurrent.*;
 +import java.util.concurrent.atomic.*;
 +
 +import static java.util.concurrent.TimeUnit.*;
 +import static org.apache.ignite.cache.CacheAtomicityMode.*;
 +import static org.apache.ignite.cache.CacheDistributionMode.*;
 +import static org.apache.ignite.cache.CacheMode.*;
 +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
 +import static org.apache.ignite.events.EventType.*;
 +
 +/**
 + *
 + */
 +public class IgniteDataStreamerProcessorSelfTest extends GridCommonAbstractTest {
 +    /** */
 +    private static ConcurrentHashMap<Object, Object> storeMap;
 +
 +    /** */
 +    private TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
 +
 +    /** */
 +    private CacheMode mode = PARTITIONED;
 +
 +    /** */
 +    private boolean nearEnabled = true;
 +
 +    /** */
 +    private boolean useCache;
 +
 +    /** */
 +    private TestStore store;
 +
 +    /** {@inheritDoc} */
 +    @Override public void afterTest() throws Exception {
 +        super.afterTest();
 +
 +        useCache = false;
 +    }
 +
 +    /** {@inheritDoc} */
 +    @SuppressWarnings({"IfMayBeConditional", "unchecked"})
 +    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
 +        IgniteConfiguration cfg = super.getConfiguration(gridName);
 +
 +        TcpDiscoverySpi spi = new TcpDiscoverySpi();
 +
 +        spi.setIpFinder(ipFinder);
 +
 +        cfg.setDiscoverySpi(spi);
 +
 +        cfg.setIncludeProperties();
 +
 +        cfg.setMarshaller(new OptimizedMarshaller(false));
 +
 +        if (useCache) {
 +            CacheConfiguration cc = defaultCacheConfiguration();
 +
 +            cc.setCacheMode(mode);
 +            cc.setAtomicityMode(TRANSACTIONAL);
 +            cc.setDistributionMode(nearEnabled ? NEAR_PARTITIONED : PARTITIONED_ONLY);
 +            cc.setWriteSynchronizationMode(FULL_SYNC);
 +
 +            cc.setEvictSynchronized(false);
 +            cc.setEvictNearSynchronized(false);
 +
 +            if (store != null) {
 +                cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
 +                cc.setReadThrough(true);
 +                cc.setWriteThrough(true);
 +            }
 +
 +            cfg.setCacheConfiguration(cc);
 +        }
 +        else
 +            cfg.setCacheConfiguration();
 +
 +        return cfg;
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testPartitioned() throws Exception {
 +        mode = PARTITIONED;
 +
 +        checkDataStreamer();
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testColocated() throws Exception {
 +        mode = PARTITIONED;
 +        nearEnabled = false;
 +
 +        checkDataStreamer();
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testReplicated() throws Exception {
 +        mode = REPLICATED;
 +
 +        checkDataStreamer();
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testLocal() throws Exception {
 +        mode = LOCAL;
 +
 +        try {
 +            checkDataStreamer();
 +
 +            assert false;
 +        }
 +        catch (IgniteCheckedException e) {
 +            // Cannot load local cache configured remotely.
 +            info("Caught expected exception: " + e);
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    @SuppressWarnings("ErrorNotRethrown")
 +    private void checkDataStreamer() throws Exception {
 +        try {
 +            Ignite g1 = startGrid(1);
 +
 +            useCache = true;
 +
 +            Ignite g2 = startGrid(2);
 +            startGrid(3);
 +
 +            final IgniteDataStreamer<Integer, Integer> ldr = g1.dataStreamer(null);
 +
 +            ldr.updater(IgniteDataStreamerCacheUpdaters.<Integer, Integer>batchedSorted());
 +
 +            final AtomicInteger idxGen = new AtomicInteger();
 +            final int cnt = 400;
 +            final int threads = 10;
 +
 +            final CountDownLatch l1 = new CountDownLatch(threads);
 +
 +            IgniteInternalFuture<?> f1 = multithreadedAsync(new Callable<Object>()
{
 +                @Override public Object call() throws Exception {
 +                    Collection<IgniteFuture<?>> futs = new ArrayList<>(cnt);
 +
 +                    for (int i = 0; i < cnt; i++) {
 +                        int idx = idxGen.getAndIncrement();
 +
 +                        futs.add(ldr.addData(idx, idx));
 +                    }
 +
 +                    l1.countDown();
 +
 +                    for (IgniteFuture<?> fut : futs)
 +                        fut.get();
 +
 +                    return null;
 +                }
 +            }, threads);
 +
 +            l1.await();
 +
 +            // This will wait until data streamer finishes loading.
 +            stopGrid(getTestGridName(1), false);
 +
 +            f1.get();
 +
 +            int s2 = internalCache(2).primaryKeySet().size();
 +            int s3 = internalCache(3).primaryKeySet().size();
 +            int total = threads * cnt;
 +
 +            assertEquals(total, s2 + s3);
 +
 +            final IgniteDataStreamer<Integer, Integer> rmvLdr = g2.dataStreamer(null);
 +
 +            rmvLdr.updater(IgniteDataStreamerCacheUpdaters.<Integer, Integer>batchedSorted());
 +
 +            final CountDownLatch l2 = new CountDownLatch(threads);
 +
 +            IgniteInternalFuture<?> f2 = multithreadedAsync(new Callable<Object>()
{
 +                @Override public Object call() throws Exception {
 +                    Collection<IgniteFuture<?>> futs = new ArrayList<>(cnt);
 +
 +                    for (int i = 0; i < cnt; i++) {
 +                        final int key = idxGen.decrementAndGet();
 +
 +                        futs.add(rmvLdr.removeData(key));
 +                    }
 +
 +                    l2.countDown();
 +
 +                    for (IgniteFuture<?> fut : futs)
 +                        fut.get();
 +
 +                    return null;
 +                }
 +            }, threads);
 +
 +            l2.await();
 +
 +            rmvLdr.close(false);
 +
 +            f2.get();
 +
 +            s2 = internalCache(2).primaryKeySet().size();
 +            s3 = internalCache(3).primaryKeySet().size();
 +
 +            assert s2 == 0 && s3 == 0 : "Incorrect entries count [s2=" + s2 + ",
s3=" + s3 + ']';
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testPartitionedIsolated() throws Exception {
 +        mode = PARTITIONED;
 +
 +        checkIsolatedDataStreamer();
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testReplicatedIsolated() throws Exception {
 +        mode = REPLICATED;
 +
 +        checkIsolatedDataStreamer();
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    private void checkIsolatedDataStreamer() throws Exception {
 +        try {
 +            useCache = true;
 +
 +            Ignite g1 = startGrid(0);
 +            startGrid(1);
 +            startGrid(2);
 +
 +            awaitPartitionMapExchange();
 +
-             GridCache<Integer, Integer> cache = ((IgniteKernal)grid(0)).cache(null);
++            IgniteCache<Integer, Integer> cache = grid(0).jcache(null);
 +
 +            for (int i = 0; i < 100; i++)
 +                cache.put(i, -1);
 +
 +            final int cnt = 40_000;
 +            final int threads = 10;
 +
 +            try (final IgniteDataStreamer<Integer, Integer> ldr = g1.dataStreamer(null))
{
 +                final AtomicInteger idxGen = new AtomicInteger();
 +
 +                IgniteInternalFuture<?> f1 = multithreadedAsync(new Callable<Object>()
{
 +                    @Override public Object call() throws Exception {
 +                        for (int i = 0; i < cnt; i++) {
 +                            int idx = idxGen.getAndIncrement();
 +
 +                            ldr.addData(idx, idx);
 +                        }
 +
 +                        return null;
 +                    }
 +                }, threads);
 +
 +                f1.get();
 +            }
 +
 +            for (int g = 0; g < 3; g++) {
 +                ClusterNode locNode = grid(g).localNode();
 +
 +                GridCacheAdapter<Integer, Integer> cache0 = ((IgniteKernal)grid(g)).internalCache(null);
 +
 +                if (cache0.isNear())
 +                    cache0 = ((GridNearCacheAdapter<Integer, Integer>)cache0).dht();
 +
 +                CacheAffinity<Integer> aff = cache0.affinity();
 +
 +                for (int key = 0; key < cnt * threads; key++) {
 +                    if (aff.isPrimary(locNode, key) || aff.isBackup(locNode, key)) {
-                         GridCacheEntryEx<Integer, Integer> entry = cache0.peekEx(key);
++                        GridCacheEntryEx entry = cache0.peekEx(key);
 +
 +                        assertNotNull("Missing entry for key: " + key, entry);
-                         assertEquals((Integer)(key < 100 ? -1 : key), entry.rawGetOrUnmarshal(false));
++                        assertEquals((key < 100 ? -1 : key),
++                            CU.value(entry.rawGetOrUnmarshal(false), cache0.context(), false));
 +                    }
 +                }
 +            }
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * Test primitive arrays can be passed into data streamer.
 +     *
 +     * @throws Exception If failed.
 +     */
 +    public void testPrimitiveArrays() throws Exception {
 +        try {
 +            useCache = true;
 +            mode = PARTITIONED;
 +
 +            Ignite g1 = startGrid(1);
 +            startGrid(2); // Reproduced only for several nodes in topology (if marshalling
is used).
 +
 +            List<Object> arrays = Arrays.<Object>asList(
 +                new byte[] {1}, new boolean[] {true, false}, new char[] {2, 3}, new short[]
{3, 4},
 +                new int[] {4, 5}, new long[] {5, 6}, new float[] {6, 7}, new double[] {7,
8});
 +
 +            IgniteDataStreamer<Object, Object> dataLdr = g1.dataStreamer(null);
 +
 +            for (int i = 0, size = arrays.size(); i < 1000; i++) {
 +                Object arr = arrays.get(i % size);
 +
 +                dataLdr.addData(i, arr);
 +                dataLdr.addData(i, fixedClosure(arr));
 +            }
 +
 +            dataLdr.close(false);
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testReplicatedMultiThreaded() throws Exception {
 +        mode = REPLICATED;
 +
 +        checkLoaderMultithreaded(1, 2);
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testPartitionedMultiThreaded() throws Exception {
 +        mode = PARTITIONED;
 +
 +        checkLoaderMultithreaded(1, 3);
 +    }
 +
 +    /**
 +     * Tests loader in multithreaded environment with various count of grids started.
 +     *
 +     * @param nodesCntNoCache How many nodes should be started without cache.
 +     * @param nodesCntCache How many nodes should be started with cache.
 +     * @throws Exception If failed.
 +     */
 +    protected void checkLoaderMultithreaded(int nodesCntNoCache, int nodesCntCache)
 +        throws Exception {
 +        try {
 +            // Start all required nodes.
 +            int idx = 1;
 +
 +            for (int i = 0; i < nodesCntNoCache; i++)
 +                startGrid(idx++);
 +
 +            useCache = true;
 +
 +            for (int i = 0; i < nodesCntCache; i++)
 +                startGrid(idx++);
 +
 +            Ignite g1 = grid(1);
 +
 +            // Get and configure loader.
 +            final IgniteDataStreamer<Integer, Integer> ldr = g1.dataStreamer(null);
 +
 +            ldr.updater(IgniteDataStreamerCacheUpdaters.<Integer, Integer>individual());
 +            ldr.perNodeBufferSize(2);
 +
 +            // Define count of puts.
 +            final AtomicInteger idxGen = new AtomicInteger();
 +
 +            final AtomicBoolean done = new AtomicBoolean();
 +
 +            try {
 +                final int totalPutCnt = 50000;
 +
 +                IgniteInternalFuture<?> fut1 = multithreadedAsync(new Callable<Object>()
{
 +                    @Override public Object call() throws Exception {
 +                        Collection<IgniteFuture<?>> futs = new ArrayList<>();
 +
 +                        while (!done.get()) {
 +                            int idx = idxGen.getAndIncrement();
 +
 +                            if (idx >= totalPutCnt) {
 +                                info(">>> Stopping producer thread since maximum
count of puts reached.");
 +
 +                                break;
 +                            }
 +
 +                            futs.add(ldr.addData(idx, idx));
 +                        }
 +
 +                        ldr.flush();
 +
 +                        for (IgniteFuture<?> fut : futs)
 +                            fut.get();
 +
 +                        return null;
 +                    }
 +                }, 5, "producer");
 +
 +                IgniteInternalFuture<?> fut2 = multithreadedAsync(new Callable<Object>()
{
 +                    @Override public Object call() throws Exception {
 +                        while (!done.get()) {
 +                            ldr.flush();
 +
 +                            U.sleep(100);
 +                        }
 +
 +                        return null;
 +                    }
 +                }, 1, "flusher");
 +
 +                // Define index of node being restarted.
 +                final int restartNodeIdx = nodesCntCache + nodesCntNoCache + 1;
 +
 +                IgniteInternalFuture<?> fut3 = multithreadedAsync(new Callable<Object>()
{
 +                    @Override public Object call() throws Exception {
 +                        try {
 +                            for (int i = 0; i < 5; i++) {
 +                                Ignite g = startGrid(restartNodeIdx);
 +
 +                                UUID id = g.cluster().localNode().id();
 +
 +                                info(">>>>>>> Started node: " + id);
 +
 +                                U.sleep(1000);
 +
 +                                stopGrid(getTestGridName(restartNodeIdx), true);
 +
 +                                info(">>>>>>> Stopped node: " + id);
 +                            }
 +                        }
 +                        finally {
 +                            done.set(true);
 +
 +                            info("Start stop thread finished.");
 +                        }
 +
 +                        return null;
 +                    }
 +                }, 1, "start-stop-thread");
 +
 +                fut1.get();
 +                fut2.get();
 +                fut3.get();
 +            }
 +            finally {
 +                ldr.close(false);
 +            }
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testLoaderApi() throws Exception {
 +        useCache = true;
 +
 +        try {
 +            Ignite g1 = startGrid(1);
 +
 +            IgniteDataStreamer<Object, Object> ldr = g1.dataStreamer(null);
 +
 +            ldr.close(false);
 +
 +            try {
 +                ldr.addData(0, 0);
 +
 +                assert false;
 +            }
 +            catch (IllegalStateException e) {
 +                info("Caught expected exception: " + e);
 +            }
 +
 +            assert ldr.future().isDone();
 +
 +            ldr.future().get();
 +
 +            try {
 +                // Create another loader.
 +                ldr = g1.dataStreamer("UNKNOWN_CACHE");
 +
 +                assert false;
 +            }
 +            catch (IllegalStateException e) {
 +                info("Caught expected exception: " + e);
 +            }
 +
 +            ldr.close(true);
 +
 +            assert ldr.future().isDone();
 +
 +            ldr.future().get();
 +
 +            // Create another loader.
 +            ldr = g1.dataStreamer(null);
 +
 +            // Cancel with future.
 +            ldr.future().cancel();
 +
 +            try {
 +                ldr.addData(0, 0);
 +
 +                assert false;
 +            }
 +            catch (IllegalStateException e) {
 +                info("Caught expected exception: " + e);
 +            }
 +
 +            assert ldr.future().isDone();
 +
 +            try {
 +                ldr.future().get();
 +
 +                assert false;
 +            }
 +            catch (IgniteFutureCancelledException e) {
 +                info("Caught expected exception: " + e);
 +            }
 +
 +            // Create another loader.
 +            ldr = g1.dataStreamer(null);
 +
 +            // This will close loader.
 +            stopGrid(getTestGridName(1), false);
 +
 +            try {
 +                ldr.addData(0, 0);
 +
 +                assert false;
 +            }
 +            catch (IllegalStateException e) {
 +                info("Caught expected exception: " + e);
 +            }
 +
 +            assert ldr.future().isDone();
 +
 +            ldr.future().get();
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * Wraps integer to closure returning it.
 +     *
 +     * @param i Value to wrap.
 +     * @return Callable.
 +     */
 +    private static Callable<Integer> callable(@Nullable final Integer i) {
 +        return new Callable<Integer>() {
 +            @Override public Integer call() throws Exception {
 +                return i;
 +            }
 +        };
 +    }
 +
 +    /**
 +     * Wraps integer to closure returning it.
 +     *
 +     * @param i Value to wrap.
 +     * @return Closure.
 +     */
 +    private static IgniteClosure<Integer, Integer> closure(@Nullable final Integer
i) {
 +        return new IgniteClosure<Integer, Integer>() {
 +            @Override public Integer apply(Integer e) {
 +                return e == null ? i : e + i;
 +            }
 +        };
 +    }
 +
 +    /**
 +     * Wraps object to closure returning it.
 +     *
 +     * @param obj Value to wrap.
 +     * @return Closure.
 +     */
 +    private static <T> IgniteClosure<T, T> fixedClosure(@Nullable final T obj)
{
 +        return new IgniteClosure<T, T>() {
 +            @Override public T apply(T e) {
 +                assert e == null || obj == null || e.getClass() == obj.getClass() :
 +                    "Expects the same types [e=" + e + ", obj=" + obj + ']';
 +
 +                return obj;
 +            }
 +        };
 +    }
 +
 +    /**
 +     * Wraps integer to closure expecting it and returning {@code null}.
 +     *
 +     * @param exp Expected closure value.
 +     * @return Remove expected cache value closure.
 +     */
 +    private static <T> IgniteClosure<T, T> removeClosure(@Nullable final T exp)
{
 +        return new IgniteClosure<T, T>() {
 +            @Override public T apply(T act) {
 +                if (exp == null ? act == null : exp.equals(act))
 +                    return null;
 +
 +                throw new AssertionError("Unexpected value [exp=" + exp + ", act=" + act
+ ']');
 +            }
 +        };
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testFlush() throws Exception {
 +        mode = LOCAL;
 +
 +        useCache = true;
 +
 +        try {
 +            Ignite g = startGrid();
 +
 +            final IgniteCache<Integer, Integer> c = g.jcache(null);
 +
 +            final IgniteDataStreamer<Integer, Integer> ldr = g.dataStreamer(null);
 +
 +            ldr.perNodeBufferSize(10);
 +
 +            for (int i = 0; i < 9; i++)
 +                ldr.addData(i, i);
 +
 +            assertTrue(c.localSize() == 0);
 +
 +            multithreaded(new Callable<Void>() {
 +                @Override
 +                public Void call() throws Exception {
 +                    ldr.flush();
 +
 +                    assertEquals(9, c.size());
 +
 +                    return null;
 +                }
 +            }, 5, "flush-checker");
 +
 +            ldr.addData(100, 100);
 +
 +            ldr.flush();
 +
 +            assertEquals(10, c.size());
 +
 +            ldr.addData(200, 200);
 +
 +            ldr.close(false);
 +
 +            ldr.future().get();
 +
 +            assertEquals(11, c.size());
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testTryFlush() throws Exception {
 +        mode = LOCAL;
 +
 +        useCache = true;
 +
 +        try {
 +            Ignite g = startGrid();
 +
 +            IgniteCache<Integer, Integer> c = g.jcache(null);
 +
 +            IgniteDataStreamer<Integer, Integer> ldr = g.dataStreamer(null);
 +
 +            ldr.perNodeBufferSize(10);
 +
 +            for (int i = 0; i < 9; i++)
 +                ldr.addData(i, i);
 +
 +            assertTrue(c.localSize() == 0);
 +
 +            ldr.tryFlush();
 +
 +            Thread.sleep(100);
 +
 +            assertEquals(9, c.size());
 +
 +            ldr.close(false);
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testFlushTimeout() throws Exception {
 +        mode = LOCAL;
 +
 +        useCache = true;
 +
 +        try {
 +            Ignite g = startGrid();
 +
 +            final CountDownLatch latch = new CountDownLatch(9);
 +
 +            g.events().localListen(new IgnitePredicate<Event>() {
 +                @Override public boolean apply(Event evt) {
 +                    latch.countDown();
 +
 +                    return true;
 +                }
 +            }, EVT_CACHE_OBJECT_PUT);
 +
 +            IgniteCache<Integer, Integer> c = g.jcache(null);
 +
 +            assertTrue(c.localSize() == 0);
 +
 +            IgniteDataStreamer<Integer, Integer> ldr = g.dataStreamer(null);
 +
 +            ldr.perNodeBufferSize(10);
 +            ldr.autoFlushFrequency(3000);
 +            ldr.allowOverwrite(true);
 +
 +            for (int i = 0; i < 9; i++)
 +                ldr.addData(i, i);
 +
 +            assertTrue(c.localSize() == 0);
 +
 +            assertFalse(latch.await(1000, MILLISECONDS));
 +
 +            assertTrue(c.localSize() == 0);
 +
 +            assertTrue(latch.await(3000, MILLISECONDS));
 +
 +            assertEquals(9, c.size());
 +
 +            ldr.close(false);
 +        }
 +        finally {
 +            stopAllGrids();
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testUpdateStore() throws Exception {
 +        storeMap = new ConcurrentHashMap<>();
 +
 +        try {
 +            store = new TestStore();
 +
 +            useCache = true;
 +
 +            Ignite ignite = startGrid(1);
 +
 +            startGrid(2);
 +            startGrid(3);
 +
 +            for (int i = 0; i < 1000; i++)
 +                storeMap.put(i, i);
 +
 +            try (IgniteDataStreamer<Object, Object> ldr = ignite.dataStreamer(null))
{
 +                ldr.allowOverwrite(true);
 +
 +                assertFalse(ldr.skipStore());
 +
 +                for (int i = 0; i < 1000; i++)
 +                    ldr.removeData(i);
 +
 +                for (int i = 1000; i < 2000; i++)
 +                    ldr.addData(i, i);
 +            }
 +
 +            for (int i = 0; i < 1000; i++)
 +                assertNull(storeMap.get(i));
 +
 +            for (int i = 1000; i < 2000; i++)
 +                assertEquals(i, storeMap.get(i));
 +
 +            try (IgniteDataStreamer<Object, Object> ldr = ignite.dataStreamer(null))
{
 +                ldr.allowOverwrite(true);
 +
 +                ldr.skipStore(true);
 +
 +                for (int i = 0; i < 1000; i++)
 +                    ldr.addData(i, i);
 +
 +                for (int i = 1000; i < 2000; i++)
 +                    ldr.removeData(i);
 +            }
 +
 +            IgniteCache<Object, Object> cache = ignite.jcache(null);
 +
 +            for (int i = 0; i < 1000; i++) {
 +                assertNull(storeMap.get(i));
 +
 +                assertEquals(i, cache.get(i));
 +            }
 +
 +            for (int i = 1000; i < 2000; i++) {
 +                assertEquals(i, storeMap.get(i));
 +
 +                assertNull(cache.localPeek(i, CachePeekMode.ONHEAP));
 +            }
 +        }
 +        finally {
 +            storeMap = null;
 +        }
 +    }
 +
 +    /**
++     * @throws Exception If failed.
++     */
++    public void testCustomUserUpdater() throws Exception {
++        useCache = true;
++
++        try {
++            Ignite ignite = startGrid(1);
++
++            startGrid(2);
++            startGrid(3);
++
++            try (IgniteDataLoader<String, TestObject> ldr = ignite.dataLoader(null))
{
++                ldr.allowOverwrite(true);
++
++                ldr.updater(new IgniteDataLoader.Updater<String, TestObject>() {
++                    @Override public void update(IgniteCache<String, TestObject> cache,
++                        Collection<Map.Entry<String, TestObject>> entries) {
++                        for (Map.Entry<String, TestObject> e : entries) {
++                            assertTrue(e.getKey() instanceof String);
++                            assertTrue(e.getValue() instanceof TestObject);
++
++                            cache.put(e.getKey(), new TestObject(e.getValue().val + 1));
++                        }
++                    }
++                });
++
++                for (int i = 0; i < 100; i++)
++                    ldr.addData(String.valueOf(i), new TestObject(i));
++            }
++
++            IgniteCache<String, TestObject> cache = ignite.jcache(null);
++
++            for (int i = 0; i < 100; i++) {
++                TestObject val = cache.get(String.valueOf(i));
++
++                assertNotNull(val);
++                assertEquals(i + 1, val.val);
++            }
++        }
++        finally {
++            stopAllGrids();
++        }
++    }
++
++    /**
 +     *
 +     */
 +    private static class TestObject {
 +        /** Value. */
 +        private final int val;
 +
 +        /**
 +         * @param val Value.
 +         */
 +        private TestObject(int val) {
 +            this.val = val;
 +        }
 +
 +        /** {@inheritDoc} */
 +        @Override public boolean equals(Object o) {
 +            if (this == o)
 +                return true;
 +
 +            if (o == null || getClass() != o.getClass())
 +                return false;
 +
 +            TestObject obj = (TestObject)o;
 +
 +            return val == obj.val;
 +        }
 +
 +        /** {@inheritDoc} */
 +        @Override public int hashCode() {
 +            return val;
 +        }
 +    }
 +
 +    /**
 +     *
 +     */
 +    private static class TestStore extends CacheStoreAdapter<Object, Object> {
 +        /** {@inheritDoc} */
 +        @Nullable @Override public Object load(Object key) {
 +            return storeMap.get(key);
 +        }
 +
 +        /** {@inheritDoc} */
 +        @Override public void write(Cache.Entry<?, ?> entry) {
 +            storeMap.put(entry.getKey(), entry.getValue());
 +        }
 +
 +        /** {@inheritDoc} */
 +        @Override public void delete(Object key) {
 +            storeMap.remove(key);
 +        }
 +    }
 +}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/00fd3c3c/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
----------------------------------------------------------------------


Mime
View raw message