| /* |
| * 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.distributed; |
| |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.UUID; |
| import java.util.concurrent.CyclicBarrier; |
| import java.util.concurrent.atomic.AtomicInteger; |
| 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.CacheRebalanceMode; |
| import org.apache.ignite.cache.eviction.lru.LruEvictionPolicy; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.internal.util.future.GridCompoundFuture; |
| 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.GridTestUtils.SF; |
| import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; |
| import org.apache.ignite.transactions.Transaction; |
| import org.apache.ignite.transactions.TransactionConcurrency; |
| import org.junit.Test; |
| |
| import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC; |
| import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; |
| import static org.apache.ignite.cache.CacheRebalanceMode.ASYNC; |
| import static org.apache.ignite.cache.CacheRebalanceMode.SYNC; |
| import static org.apache.ignite.configuration.IgniteConfiguration.DFLT_REBALANCE_BATCH_SIZE; |
| import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC; |
| import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ; |
| |
| /** |
| * Test node restart. |
| */ |
| public abstract class GridCacheAbstractNodeRestartSelfTest extends GridCommonAbstractTest { |
| /** Cache name. */ |
| protected static final String CACHE_NAME = "TEST_CACHE"; |
| |
| /** */ |
| private static final long TEST_TIMEOUT = 5 * 60 * 1000; |
| |
| /** Default backups. */ |
| private static final int DFLT_BACKUPS = 1; |
| |
| /** Partitions. */ |
| private static final int DFLT_PARTITIONS = 521; |
| |
| /** Preload batch size. */ |
| private static final int DFLT_BATCH_SIZE = DFLT_REBALANCE_BATCH_SIZE; |
| |
| /** Number of key backups. Each test method can set this value as required. */ |
| protected int backups = DFLT_BACKUPS; |
| |
| /** */ |
| private static final int DFLT_NODE_CNT = 4; |
| |
| /** */ |
| private static final int DFLT_KEY_CNT = 100; |
| |
| /** */ |
| private static final int DFLT_RETRIES = 10; |
| |
| /** */ |
| private static final int LOG_FREQ = 1000; |
| |
| /** */ |
| private static final Random RAND = new Random(); |
| |
| /** */ |
| private static volatile int idx = -1; |
| |
| /** Preload mode. */ |
| protected CacheRebalanceMode rebalancMode = ASYNC; |
| |
| /** */ |
| protected boolean evict = false; |
| |
| /** */ |
| protected int rebalancBatchSize = DFLT_BATCH_SIZE; |
| |
| /** Number of partitions. */ |
| protected int partitions = DFLT_PARTITIONS; |
| |
| /** Node count. */ |
| protected int nodeCnt = DFLT_NODE_CNT; |
| |
| /** Key count. */ |
| protected int keyCnt = DFLT_KEY_CNT; |
| |
| /** Retries. */ |
| private int retries = DFLT_RETRIES; |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { |
| IgniteConfiguration c = super.getConfiguration(igniteInstanceName); |
| |
| ((TcpCommunicationSpi)c.getCommunicationSpi()).setSharedMemoryPort(-1); |
| |
| // Discovery. |
| TcpDiscoverySpi disco = (TcpDiscoverySpi)c.getDiscoverySpi(); |
| |
| disco.setSocketTimeout(30_000); |
| disco.setAckTimeout(30_000); |
| disco.setNetworkTimeout(30_000); |
| |
| CacheConfiguration ccfg = cacheConfiguration(); |
| |
| if (evict) { |
| LruEvictionPolicy plc = new LruEvictionPolicy(); |
| |
| plc.setMaxSize(100); |
| |
| ccfg.setEvictionPolicy(plc); |
| ccfg.setOnheapCacheEnabled(true); |
| } |
| |
| c.setCacheConfiguration(ccfg); |
| |
| return c; |
| } |
| |
| /** |
| * @return Cache configuration. |
| */ |
| protected abstract CacheConfiguration cacheConfiguration(); |
| |
| /** {@inheritDoc} */ |
| @Override protected void afterTest() throws Exception { |
| super.afterTest(); |
| |
| stopAllGrids(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTest() throws Exception { |
| backups = DFLT_BACKUPS; |
| partitions = DFLT_PARTITIONS; |
| rebalancMode = ASYNC; |
| evict = false; |
| rebalancBatchSize = DFLT_BATCH_SIZE; |
| nodeCnt = DFLT_NODE_CNT; |
| keyCnt = DFLT_KEY_CNT; |
| retries = DFLT_RETRIES; |
| idx = -1; |
| } |
| |
| /** |
| * @return Cache atomicity mode. |
| */ |
| protected CacheAtomicityMode atomicityMode() { |
| return TRANSACTIONAL; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected long getTestTimeout() { |
| return TEST_TIMEOUT; |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| private void startGrids() throws Exception { |
| for (int i = 0; i < nodeCnt; i++) { |
| startGrid(i); |
| |
| if (idx < 0) |
| idx = i; |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestart() throws Exception { |
| rebalancMode = SYNC; |
| partitions = 3; |
| nodeCnt = 2; |
| keyCnt = 10; |
| retries = 3; |
| |
| info("*** STARTING TEST ***"); |
| |
| startGrids(); |
| |
| try { |
| IgniteCache<Integer, String> c = grid(idx).cache(CACHE_NAME); |
| |
| for (int j = 0; j < retries; j++) { |
| for (int i = 0; i < keyCnt; i++) |
| c.put(i, Integer.toString(i)); |
| |
| info("Stored items."); |
| |
| checkGet(c, j); |
| |
| info("Stopping node: " + idx); |
| |
| stopGrid(idx); |
| |
| info("Starting node: " + idx); |
| |
| Ignite ignite = startGrid(idx); |
| |
| c = ignite.cache(CACHE_NAME); |
| |
| checkGet(c, j); |
| } |
| } |
| finally { |
| stopAllGrids(true); |
| } |
| } |
| |
| /** |
| * @param c Cache. |
| * @param attempt Attempt. |
| * @throws Exception If failed. |
| */ |
| private void checkGet(IgniteCache<Integer, String> c, int attempt) throws Exception { |
| for (int i = 0; i < keyCnt; i++) { |
| String v = c.get(i); |
| |
| if (v == null) { |
| printFailureDetails(c, i, attempt); |
| |
| fail("Value is null [key=" + i + ", attempt=" + attempt + "]"); |
| } |
| |
| if (!Integer.toString(i).equals(v)) { |
| printFailureDetails(c, i, attempt); |
| |
| fail("Wrong value for key [key=" + |
| i + ", actual value=" + v + ", expected value=" + Integer.toString(i) + "]"); |
| } |
| } |
| |
| info("Read items."); |
| } |
| |
| /** |
| * @return Transaction concurrency to use in tests. |
| */ |
| protected TransactionConcurrency txConcurrency() { |
| return PESSIMISTIC; |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutTwoNodesNoBackups() throws Throwable { |
| backups = 0; |
| nodeCnt = 2; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 3_000); |
| |
| checkRestartWithPut(duration, 1, 1); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxTwoNodesNoBackups() throws Throwable { |
| backups = 0; |
| nodeCnt = 2; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 3_000); |
| |
| checkRestartWithTx(duration, 1, 1, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutTwoNodesOneBackup() throws Throwable { |
| backups = 1; |
| nodeCnt = 2; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 3_000); |
| |
| checkRestartWithPut(duration, 1, 1); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxTwoNodesOneBackup() throws Throwable { |
| backups = 1; |
| nodeCnt = 2; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 3_000); |
| |
| checkRestartWithTx(duration, 1, 1, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutFourNodesNoBackups() throws Throwable { |
| backups = 0; |
| nodeCnt = 4; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 5_000); |
| |
| checkRestartWithPut(duration, 2, 2); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxFourNodesNoBackups() throws Throwable { |
| backups = 0; |
| nodeCnt = 4; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 5_000); |
| |
| checkRestartWithTx(duration, 2, 2, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutFourNodesOneBackups() throws Throwable { |
| backups = 1; |
| nodeCnt = 4; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 5_000); |
| |
| checkRestartWithPut(duration, 2, 2); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutFourNodesOneBackupsOffheapEvict() throws Throwable { |
| backups = 1; |
| nodeCnt = 4; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = true; |
| |
| long duration = SF.applyLB(30_000, 5_000); |
| |
| checkRestartWithPut(duration, 2, 2); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxFourNodesOneBackups() throws Throwable { |
| backups = 1; |
| nodeCnt = 4; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 5_000); |
| |
| checkRestartWithTx(duration, 2, 2, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxFourNodesOneBackupsOffheapEvict() throws Throwable { |
| backups = 1; |
| nodeCnt = 4; |
| keyCnt = 100_000; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = true; |
| |
| long duration = SF.applyLB(30_000, 3_000); |
| |
| checkRestartWithTx(duration, 2, 2, 100); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutSixNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 6; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithPut(duration, 3, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxSixNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 6; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithTx(duration, 3, 3, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutEightNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 8; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithPut(duration, 4, 4); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxEightNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 8; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithTx(duration, 4, 4, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithPutTenNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 10; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithPut(duration, 5, 5); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxTenNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 10; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithTx(duration, 5, 5, 3); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxPutAllTenNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 10; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithTxPutAll(duration, 5, 5); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testRestartWithTxPutAllFourNodesTwoBackups() throws Throwable { |
| backups = 2; |
| nodeCnt = 4; |
| keyCnt = 10; |
| partitions = 29; |
| rebalancMode = ASYNC; |
| evict = false; |
| |
| long duration = SF.applyLB(30_000, 6_000); |
| |
| checkRestartWithTxPutAll(duration, 2, 2); |
| } |
| |
| /** |
| * @param duration Test duration. |
| * @param putThreads Put threads count. |
| * @param restartThreads Restart threads count. |
| * @throws Exception If failed. |
| */ |
| private void checkRestartWithPut(long duration, int putThreads, int restartThreads) throws Throwable { |
| final long endTime = System.currentTimeMillis() + duration; |
| |
| final AtomicReference<Throwable> err = new AtomicReference<>(); |
| |
| startGrids(); |
| |
| GridCompoundFuture<Long, Long> fut = new GridCompoundFuture<>(); |
| |
| try { |
| final AtomicInteger putCntr = new AtomicInteger(); |
| final CyclicBarrier barrier = new CyclicBarrier(putThreads + restartThreads); |
| final AtomicInteger startedThreadCntr = new AtomicInteger(); |
| |
| fut.add(GridTestUtils.runMultiThreadedAsync(() -> { |
| try { |
| int gridIdx = startedThreadCntr.getAndIncrement(); |
| |
| barrier.await(); |
| |
| info("Starting put thread: " + gridIdx); |
| |
| Thread.currentThread().setName("put-worker-" + grid(gridIdx).name()); |
| |
| IgniteCache<Integer, String> cache = grid(gridIdx).cache(CACHE_NAME); |
| |
| while (System.currentTimeMillis() < endTime && err.get() == null) { |
| int key = RAND.nextInt(keyCnt); |
| |
| try { |
| cache.put(key, Integer.toString(key)); |
| } |
| catch (IgniteException | CacheException ignored) { |
| // It is ok if primary node leaves grid. |
| } |
| |
| cache.get(key); |
| |
| int c = putCntr.incrementAndGet(); |
| |
| if (c % LOG_FREQ == 0) |
| info(">>> Put iteration [cnt=" + c + ", key=" + key + ']'); |
| } |
| } |
| catch (Exception e) { |
| err.compareAndSet(null, e); |
| |
| error("Unexpected exception in put-worker.", e); |
| } |
| }, putThreads, "put-worker")); |
| |
| GridTestUtils.waitForCondition(() -> startedThreadCntr.get() == putThreads, getTestTimeout()); |
| |
| fut.add(GridTestUtils.runMultiThreadedAsync(() -> { |
| try { |
| int gridIdx = startedThreadCntr.getAndIncrement(); |
| |
| barrier.await(); |
| |
| info("Starting restart thread: " + gridIdx); |
| |
| int cnt = 0; |
| |
| while (System.currentTimeMillis() < endTime && err.get() == null) { |
| log.info(">>>>>>> Stopping grid " + gridIdx); |
| |
| stopGrid(gridIdx); |
| |
| log.info(">>>>>>> Starting grid " + gridIdx); |
| |
| startGrid(gridIdx); |
| |
| int c = ++cnt; |
| |
| if (c % LOG_FREQ == 0) |
| info(">>> Restart iteration: " + c); |
| } |
| } |
| catch (Exception e) { |
| err.compareAndSet(null, e); |
| |
| error("Unexpected exception in restart-worker.", e); |
| } |
| }, restartThreads, "restart-worker")); |
| |
| fut.markInitialized(); |
| |
| fut.get(getTestTimeout()); |
| |
| if (err.get() != null) |
| throw err.get(); |
| } |
| finally { |
| if (!fut.isDone()) |
| fut.cancel(); |
| } |
| } |
| |
| /** |
| * @param duration Test duration. |
| * @param putThreads Put threads count. |
| * @param restartThreads Restart threads count. |
| * @param txKeys Keys per transaction. |
| * @throws Exception If failed. |
| */ |
| private void checkRestartWithTx(long duration, |
| int putThreads, |
| int restartThreads, |
| final int txKeys) throws Throwable { |
| if (atomicityMode() == ATOMIC) |
| return; |
| |
| final long endTime = System.currentTimeMillis() + duration; |
| final AtomicReference<Throwable> err = new AtomicReference<>(); |
| |
| startGrids(); |
| |
| GridCompoundFuture<Long, Long> fut = new GridCompoundFuture<>(); |
| |
| try { |
| final AtomicInteger txCntr = new AtomicInteger(); |
| final CyclicBarrier barrier = new CyclicBarrier(putThreads + restartThreads); |
| final AtomicInteger startedThreadCntr = new AtomicInteger(); |
| |
| fut.add(GridTestUtils.runMultiThreadedAsync(() -> { |
| try { |
| int gridIdx = startedThreadCntr.getAndIncrement(); |
| |
| barrier.await(); |
| |
| info("Starting put thread: " + gridIdx); |
| |
| Ignite ignite = grid(gridIdx); |
| |
| Thread.currentThread().setName("put-worker-" + ignite.name()); |
| |
| UUID locNodeId = ignite.cluster().localNode().id(); |
| |
| IgniteCache<Integer, String> cache = ignite.cache(CACHE_NAME).withAllowAtomicOpsInTx(); |
| |
| List<Integer> keys = new ArrayList<>(txKeys); |
| |
| while (System.currentTimeMillis() < endTime && err.get() == null) { |
| keys.clear(); |
| |
| for (int i = 0; i < txKeys; i++) |
| keys.add(RAND.nextInt(keyCnt)); |
| |
| // Ensure lock order. |
| Collections.sort(keys); |
| |
| int c = 0; |
| |
| try { |
| IgniteTransactions txs = ignite.transactions(); |
| |
| try (Transaction tx = txs.txStart(txConcurrency(), REPEATABLE_READ)) { |
| c = txCntr.incrementAndGet(); |
| |
| if (c % LOG_FREQ == 0) { |
| info(">>> Tx iteration started [cnt=" + c + |
| ", keys=" + keys + |
| ", locNodeId=" + locNodeId + ']'); |
| } |
| |
| for (int key : keys) { |
| int op = cacheOp(); |
| |
| if (op == 1) |
| cache.put(key, Integer.toString(key)); |
| else if (op == 2) |
| cache.remove(key); |
| else |
| cache.get(key); |
| } |
| |
| tx.commit(); |
| } |
| } |
| catch (IgniteException | CacheException ignored) { |
| // It is ok if primary node leaves grid. |
| } |
| |
| if (c % LOG_FREQ == 0) { |
| info(">>> Tx iteration finished [cnt=" + c + |
| ", cacheSize=" + cache.localSize() + |
| ", keys=" + keys + |
| ", locNodeId=" + locNodeId + ']'); |
| } |
| } |
| |
| info(">>> " + Thread.currentThread().getName() + " finished."); |
| } |
| catch (Exception e) { |
| err.compareAndSet(null, e); |
| |
| error("Unexpected exception in put-worker.", e); |
| } |
| }, putThreads, "put-worker")); |
| |
| GridTestUtils.waitForCondition(() -> startedThreadCntr.get() == putThreads, getTestTimeout()); |
| |
| fut.add(GridTestUtils.runMultiThreadedAsync(() -> { |
| try { |
| int gridIdx = startedThreadCntr.incrementAndGet(); |
| |
| barrier.await(); |
| |
| info("Starting restart thread: " + gridIdx); |
| |
| int cnt = 0; |
| |
| while (System.currentTimeMillis() < endTime && err.get() == null) { |
| stopGrid(getTestIgniteInstanceName(gridIdx), false, false); |
| startGrid(gridIdx); |
| |
| int c = ++cnt; |
| |
| if (c % LOG_FREQ == 0) |
| info(">>> Restart iteration: " + c); |
| } |
| |
| info(">>> " + Thread.currentThread().getName() + " finished."); |
| } |
| catch (Exception e) { |
| err.compareAndSet(null, e); |
| |
| error("Unexpected exception in restart-worker.", e); |
| } |
| }, restartThreads, "restart-worker")); |
| |
| fut.markInitialized(); |
| |
| fut.get(getTestTimeout()); |
| |
| if (err.get() != null) |
| throw err.get(); |
| } |
| finally { |
| if (!fut.isDone()) |
| fut.cancel(); |
| } |
| } |
| |
| /** |
| * @param duration Test duration. |
| * @param putThreads Put threads count. |
| * @param restartThreads Restart threads count. |
| * @throws Exception If failed. |
| */ |
| private void checkRestartWithTxPutAll(long duration, int putThreads, int restartThreads) throws Throwable { |
| if (atomicityMode() == ATOMIC) |
| return; |
| |
| final long endTime = System.currentTimeMillis() + duration; |
| |
| final AtomicReference<Throwable> err = new AtomicReference<>(); |
| |
| startGrids(); |
| |
| GridCompoundFuture<Long, Long> fut = new GridCompoundFuture<>(); |
| |
| try { |
| final AtomicInteger txCntr = new AtomicInteger(); |
| final CyclicBarrier barrier = new CyclicBarrier(putThreads + restartThreads); |
| final AtomicInteger startedThreadCntr = new AtomicInteger(); |
| final int txKeys = 3; |
| |
| fut.add(GridTestUtils.runMultiThreadedAsync(() -> { |
| try { |
| int gridIdx = startedThreadCntr.getAndIncrement(); |
| |
| barrier.await(); |
| |
| info("Starting put thread: " + gridIdx); |
| |
| Ignite ignite = grid(gridIdx); |
| |
| Thread.currentThread().setName("put-worker-" + ignite.name()); |
| |
| UUID locNodeId = ignite.cluster().localNode().id(); |
| |
| IgniteCache<Integer, String> cache = ignite.cache(CACHE_NAME).withAllowAtomicOpsInTx(); |
| |
| List<Integer> keys = new ArrayList<>(txKeys); |
| |
| while (System.currentTimeMillis() < endTime && err.get() == null) { |
| keys.clear(); |
| |
| for (int i = 0; i < txKeys; i++) |
| keys.add(RAND.nextInt(keyCnt)); |
| |
| // Ensure lock order. |
| Collections.sort(keys); |
| |
| int c = 0; |
| |
| try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| c = txCntr.incrementAndGet(); |
| |
| if (c % LOG_FREQ == 0) |
| info(">>> Tx iteration started [cnt=" + c + ", keys=" + keys + ", " + |
| "locNodeId=" + locNodeId + ']'); |
| |
| Map<Integer, String> batch = new LinkedHashMap<>(); |
| |
| for (int key : keys) |
| batch.put(key, String.valueOf(key)); |
| |
| cache.putAll(batch); |
| |
| tx.commit(); |
| } |
| catch (IgniteException | CacheException ignored) { |
| // It is ok if primary node leaves grid. |
| } |
| |
| if (c % LOG_FREQ == 0) { |
| info(">>> Tx iteration finished [cnt=" + c + |
| ", keys=" + keys + ", " + |
| "locNodeId=" + locNodeId + ']'); |
| } |
| } |
| } |
| catch (Exception e) { |
| err.compareAndSet(null, e); |
| |
| error("Unexpected exception in put-worker.", e); |
| } |
| }, putThreads, "put-worker")); |
| |
| GridTestUtils.waitForCondition(() -> startedThreadCntr.get() == putThreads, getTestTimeout()); |
| |
| fut.add(GridTestUtils.runMultiThreadedAsync(() -> { |
| try { |
| int gridIdx = startedThreadCntr.getAndIncrement(); |
| |
| barrier.await(); |
| |
| info("Starting restart thread: " + gridIdx); |
| |
| int cnt = 0; |
| |
| while (System.currentTimeMillis() < endTime && err.get() == null) { |
| stopGrid(gridIdx); |
| startGrid(gridIdx); |
| |
| int c = ++cnt; |
| |
| if (c % LOG_FREQ == 0) |
| info(">>> Restart iteration: " + c); |
| } |
| } |
| catch (Exception e) { |
| err.compareAndSet(null, e); |
| |
| error("Unexpected exception in restart-worker.", e); |
| } |
| }, restartThreads, "restart-worker")); |
| |
| fut.markInitialized(); |
| |
| fut.get(getTestTimeout()); |
| |
| if (err.get() != null) |
| throw err.get(); |
| } |
| finally { |
| if (!fut.isDone()) |
| fut.cancel(); |
| } |
| } |
| |
| /** |
| * @return Cache operation. |
| */ |
| private int cacheOp() { |
| return RAND.nextInt(3) + 1; |
| } |
| |
| /** |
| * @param c Cache projection. |
| * @param key Key. |
| * @param attempt Attempt. |
| */ |
| private void printFailureDetails(IgniteCache<Integer, String> c, int key, int attempt) { |
| Ignite ignite = c.unwrap(Ignite.class); |
| |
| error("*** Failure details ***"); |
| error("Key: " + key); |
| error("Partition: " + ignite.affinity(c.getName()).partition(key)); |
| error("Attempt: " + attempt); |
| error("Node: " + ignite.cluster().localNode().id()); |
| } |
| } |
