| /* |
| * 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.Collection; |
| import java.util.HashSet; |
| import java.util.Map; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CyclicBarrier; |
| import java.util.concurrent.ThreadLocalRandom; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.AtomicReference; |
| import javax.cache.CacheException; |
| import org.apache.ignite.Ignite; |
| import org.apache.ignite.IgniteCache; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.IgniteTransactions; |
| import org.apache.ignite.cache.CacheAtomicityMode; |
| import org.apache.ignite.cache.CacheMode; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.configuration.NearCacheConfiguration; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; |
| import org.apache.ignite.internal.util.lang.GridTuple; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.T2; |
| import org.apache.ignite.internal.util.typedef.X; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; |
| import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; |
| import org.apache.ignite.testframework.GridTestUtils; |
| import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; |
| import org.apache.ignite.transactions.Transaction; |
| import org.apache.ignite.transactions.TransactionConcurrency; |
| import org.apache.ignite.transactions.TransactionIsolation; |
| import org.junit.Test; |
| |
| import static org.apache.ignite.IgniteSystemProperties.IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE; |
| import static org.apache.ignite.cache.CacheMode.PARTITIONED; |
| import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; |
| import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC; |
| import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC; |
| import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ; |
| import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE; |
| |
| /** |
| * Tests that removes are not lost when topology changes. |
| */ |
| public abstract class GridCacheAbstractRemoveFailureTest extends GridCommonAbstractTest { |
| /** */ |
| private static final int GRID_CNT = 3; |
| |
| /** Keys count. */ |
| private static final int KEYS_CNT = 10_000; |
| |
| /** Test duration. */ |
| private static final long DUR = 90 * 1000L; |
| |
| /** Cache data assert frequency. */ |
| private static final long ASSERT_FREQ = 10_000; |
| |
| /** Kill delay. */ |
| private static final T2<Integer, Integer> KILL_DELAY = new T2<>(2000, 5000); |
| |
| /** Start delay. */ |
| private static final T2<Integer, Integer> START_DELAY = new T2<>(2000, 5000); |
| |
| /** */ |
| private static String sizePropVal; |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { |
| IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); |
| |
| ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(true); |
| |
| if (testClientNode() && getTestIgniteInstanceName(0).equals(igniteInstanceName)) |
| cfg.setClientMode(true); |
| |
| ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setSharedMemoryPort(-1); |
| |
| return cfg; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTestsStarted() throws Exception { |
| // Need to increase value set in GridAbstractTest |
| sizePropVal = System.getProperty(IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE); |
| |
| System.setProperty(IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE, "100000"); |
| |
| startGrids(GRID_CNT); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void afterTestsStopped() throws Exception { |
| System.setProperty(IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE, sizePropVal != null ? sizePropVal : ""); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected long getTestTimeout() { |
| return DUR + 60_000; |
| } |
| |
| /** |
| * @return Cache mode. |
| */ |
| protected abstract CacheMode cacheMode(); |
| |
| /** |
| * @return Cache atomicity mode. |
| */ |
| protected abstract CacheAtomicityMode atomicityMode(); |
| |
| /** |
| * @return Near cache configuration. |
| */ |
| protected abstract NearCacheConfiguration nearCache(); |
| |
| /** |
| * @return {@code True} if test updates from client node. |
| */ |
| protected boolean testClientNode() { |
| return false; |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPutAndRemove() throws Exception { |
| putAndRemove(duration(), null, null); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPutAndRemovePessimisticTx() throws Exception { |
| if (atomicityMode() != CacheAtomicityMode.TRANSACTIONAL) |
| return; |
| |
| putAndRemove(duration(), PESSIMISTIC, REPEATABLE_READ); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPutAndRemoveOptimisticSerializableTx() throws Exception { |
| if (atomicityMode() != CacheAtomicityMode.TRANSACTIONAL) |
| return; |
| |
| putAndRemove(duration(), OPTIMISTIC, SERIALIZABLE); |
| } |
| |
| /** */ |
| protected long duration() { |
| return DUR; |
| } |
| |
| /** |
| * @param duration Test duration. |
| * @param txConcurrency Transaction concurrency if test explicit transaction. |
| * @param txIsolation Transaction isolation if test explicit transaction. |
| * @throws Exception If failed. |
| */ |
| private void putAndRemove(long duration, |
| final TransactionConcurrency txConcurrency, |
| final TransactionIsolation txIsolation) throws Exception { |
| assertEquals(testClientNode(), (boolean)grid(0).configuration().isClientMode()); |
| |
| grid(0).destroyCache(DEFAULT_CACHE_NAME); |
| |
| CacheConfiguration<Integer, Integer> ccfg = new CacheConfiguration<>(DEFAULT_CACHE_NAME); |
| |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| |
| ccfg.setCacheMode(cacheMode()); |
| |
| if (cacheMode() == PARTITIONED) |
| ccfg.setBackups(1); |
| |
| ccfg.setAtomicityMode(atomicityMode()); |
| ccfg.setNearConfiguration(nearCache()); |
| |
| final IgniteCache<Integer, Integer> sndCache0 = grid(0).createCache(ccfg); |
| |
| final AtomicBoolean stop = new AtomicBoolean(); |
| |
| final AtomicLong cntr = new AtomicLong(); |
| |
| final AtomicLong errCntr = new AtomicLong(); |
| |
| // Expected values in cache. |
| final Map<Integer, GridTuple<Integer>> expVals = new ConcurrentHashMap<>(); |
| |
| final AtomicReference<CyclicBarrier> cmp = new AtomicReference<>(); |
| |
| IgniteInternalFuture<?> updateFut = GridTestUtils.runAsync(new Callable<Void>() { |
| @Override public Void call() throws Exception { |
| Thread.currentThread().setName("update-thread"); |
| |
| ThreadLocalRandom rnd = ThreadLocalRandom.current(); |
| |
| IgniteTransactions txs = sndCache0.unwrap(Ignite.class).transactions(); |
| |
| while (!stop.get()) { |
| for (int i = 0; i < 100; i++) { |
| int key = rnd.nextInt(KEYS_CNT); |
| |
| boolean put = rnd.nextInt(0, 100) > 10; |
| |
| while (true) { |
| try { |
| if (put) { |
| boolean failed = false; |
| |
| if (txConcurrency != null) { |
| try (Transaction tx = txs.txStart(txConcurrency, txIsolation)) { |
| sndCache0.put(key, i); |
| |
| tx.commit(); |
| } |
| catch (CacheException | IgniteException e) { |
| if (!X.hasCause(e, ClusterTopologyCheckedException.class)) { |
| log.error("Unexpected error: " + e); |
| |
| throw e; |
| } |
| |
| failed = true; |
| } |
| } |
| else |
| sndCache0.put(key, i); |
| |
| if (!failed) |
| expVals.put(key, F.t(i)); |
| } |
| else { |
| boolean failed = false; |
| |
| if (txConcurrency != null) { |
| try (Transaction tx = txs.txStart(txConcurrency, txIsolation)) { |
| sndCache0.remove(key); |
| |
| tx.commit(); |
| } |
| catch (CacheException | IgniteException e) { |
| if (!X.hasCause(e, ClusterTopologyCheckedException.class)) { |
| log.error("Unexpected error: " + e); |
| |
| throw e; |
| } |
| |
| failed = true; |
| } |
| } |
| else |
| sndCache0.remove(key); |
| |
| if (!failed) |
| expVals.put(key, F.<Integer>t(null)); |
| } |
| |
| break; |
| } |
| catch (CacheException e) { |
| if (put) |
| log.error("Put failed [key=" + key + ", val=" + i + ']', e); |
| else |
| log.error("Remove failed [key=" + key + ']', e); |
| |
| errCntr.incrementAndGet(); |
| } |
| } |
| } |
| |
| cntr.addAndGet(100); |
| |
| CyclicBarrier barrier = cmp.get(); |
| |
| if (barrier != null) { |
| log.info("Wait data check."); |
| |
| barrier.await(60_000, TimeUnit.MILLISECONDS); |
| |
| log.info("Finished wait data check."); |
| } |
| } |
| |
| return null; |
| } |
| }); |
| |
| IgniteInternalFuture killFut = createAndRunConcurrentAction(stop, cmp); |
| |
| try { |
| long stopTime = duration + U.currentTimeMillis(); |
| |
| long nextAssert = U.currentTimeMillis() + ASSERT_FREQ; |
| |
| while (U.currentTimeMillis() < stopTime) { |
| long start = System.nanoTime(); |
| |
| long ops = cntr.longValue(); |
| |
| U.sleep(1000); |
| |
| long diff = cntr.longValue() - ops; |
| |
| double time = (System.nanoTime() - start) / 1_000_000_000d; |
| |
| long opsPerSecond = (long)(diff / time); |
| |
| log.info("Operations/second: " + opsPerSecond); |
| |
| if (U.currentTimeMillis() >= nextAssert) { |
| CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() { |
| @Override public void run() { |
| try { |
| cmp.set(null); |
| |
| log.info("Checking cache content."); |
| |
| assertCacheContent(expVals); |
| |
| log.info("Finished check cache content."); |
| } |
| catch (Throwable e) { |
| log.error("Unexpected error: " + e, e); |
| |
| throw e; |
| } |
| } |
| }); |
| |
| log.info("Start cache content check."); |
| |
| cmp.set(barrier); |
| |
| try { |
| barrier.await(60_000, TimeUnit.MILLISECONDS); |
| } |
| catch (TimeoutException e) { |
| U.dumpThreads(log); |
| |
| fail("Failed to check cache content: " + e); |
| } |
| |
| log.info("Cache content check done."); |
| |
| nextAssert = System.currentTimeMillis() + ASSERT_FREQ; |
| } |
| } |
| } |
| finally { |
| stop.set(true); |
| } |
| |
| killFut.get(); |
| |
| updateFut.get(); |
| |
| log.info("Test finished. Update errors: " + errCntr.get()); |
| } |
| |
| /** */ |
| protected IgniteInternalFuture createAndRunConcurrentAction(final AtomicBoolean stop, final AtomicReference<CyclicBarrier> cmp) { |
| return GridTestUtils.runAsync(new Callable<Void>() { |
| @Override public Void call() throws Exception { |
| Thread.currentThread().setName("restart-thread"); |
| |
| while (!stop.get()) { |
| U.sleep(random(KILL_DELAY.get1(), KILL_DELAY.get2())); |
| |
| killAndRestart(stop, random(1, GRID_CNT + 1)); |
| |
| CyclicBarrier barrier = cmp.get(); |
| |
| if (barrier != null) { |
| log.info("Wait data check."); |
| |
| barrier.await(60_000, TimeUnit.MILLISECONDS); |
| |
| log.info("Finished wait data check."); |
| } |
| } |
| |
| return null; |
| } |
| }); |
| } |
| |
| /** |
| * @param stop Stop flag. |
| * @throws Exception If failed. |
| */ |
| protected void killAndRestart(AtomicBoolean stop, int nodeIdx) throws Exception { |
| if (stop.get()) |
| return; |
| |
| log.info("Killing node " + nodeIdx); |
| |
| stopGrid(nodeIdx); |
| |
| U.sleep(random(START_DELAY.get1(), START_DELAY.get2())); |
| |
| log.info("Restarting node " + nodeIdx); |
| |
| startGrid(nodeIdx); |
| |
| if (stop.get()) |
| return; |
| |
| U.sleep(1000); |
| } |
| |
| /** |
| * @param expVals Expected values in cache. |
| */ |
| private void assertCacheContent(Map<Integer, GridTuple<Integer>> expVals) { |
| assert !expVals.isEmpty(); |
| |
| Collection<Integer> failedKeys = new HashSet<>(); |
| |
| for (int i = 0; i < GRID_CNT; i++) { |
| Ignite ignite = grid(i); |
| |
| IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); |
| |
| for (Map.Entry<Integer, GridTuple<Integer>> expVal : expVals.entrySet()) { |
| Integer val = cache.get(expVal.getKey()); |
| |
| if (!F.eq(expVal.getValue().get(), val)) { |
| failedKeys.add(expVal.getKey()); |
| |
| boolean primary = affinity(cache).isPrimary(ignite.cluster().localNode(), expVal.getKey()); |
| boolean backup = affinity(cache).isBackup(ignite.cluster().localNode(), expVal.getKey()); |
| |
| log.error("Unexpected cache data [exp=" + expVal + |
| ", actual=" + val + |
| ", nodePrimary=" + primary + |
| ", nodeBackup=" + backup + |
| ", nodeIdx" + i + |
| ", nodeId=" + ignite.cluster().localNode().id() + ']'); |
| } |
| } |
| } |
| |
| assertTrue("Unexpected data for keys: " + failedKeys, failedKeys.isEmpty()); |
| } |
| |
| /** |
| * @param min Min possible value. |
| * @param max Max possible value (exclusive). |
| * @return Random value. |
| */ |
| protected static int random(int min, int max) { |
| if (max == min) |
| return max; |
| |
| return ThreadLocalRandom.current().nextInt(min, max); |
| } |
| } |
