diff --git a/framework/src/main/java/org/tron/common/backup/BackupManager.java b/framework/src/main/java/org/tron/common/backup/BackupManager.java index a870c183a8d..497f38544bc 100644 --- a/framework/src/main/java/org/tron/common/backup/BackupManager.java +++ b/framework/src/main/java/org/tron/common/backup/BackupManager.java @@ -58,7 +58,7 @@ public class BackupManager implements EventHandler { private MessageHandler messageHandler; @Getter - private BackupStatusEnum status = MASTER; + private volatile BackupStatusEnum status = MASTER; private volatile long lastKeepAliveTime; diff --git a/framework/src/main/java/org/tron/core/metrics/blockchain/BlockChainMetricManager.java b/framework/src/main/java/org/tron/core/metrics/blockchain/BlockChainMetricManager.java index f39cf66a8ad..f7748dc5163 100644 --- a/framework/src/main/java/org/tron/core/metrics/blockchain/BlockChainMetricManager.java +++ b/framework/src/main/java/org/tron/core/metrics/blockchain/BlockChainMetricManager.java @@ -138,10 +138,10 @@ public void applyBlock(BlockCapsule block) { BlockCapsule oldBlock = witnessInfo.get(witnessAddress); if ((!oldBlock.getBlockId().equals(block.getBlockId())) && oldBlock.getTimeStamp() == block.getTimeStamp()) { + dupWitnessBlockNum.put(witnessAddress, block.getNum()); MetricsUtil.counterInc(MetricsKey.BLOCKCHAIN_DUP_WITNESS + witnessAddress); Metrics.counterInc(MetricKeys.Counter.MINER, 1, StringUtil.encode58Check(address), MetricLabels.Counter.MINE_DUP); - dupWitnessBlockNum.put(witnessAddress, block.getNum()); } } witnessInfo.put(witnessAddress, block); @@ -204,7 +204,7 @@ private List getDupWitness() { for (Map.Entry entry : dupWitnessMap.entrySet()) { DupWitnessInfo dupWitness = new DupWitnessInfo(); String witness = entry.getKey().substring(MetricsKey.BLOCKCHAIN_DUP_WITNESS.length()); - long blockNum = dupWitnessBlockNum.get(witness); + long blockNum = dupWitnessBlockNum.getOrDefault(witness, 0L); dupWitness.setAddress(witness); dupWitness.setBlockNum(blockNum); dupWitness.setCount((int) entry.getValue().getCount()); diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index bd2e428418c..c2c6d6d4cc7 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java @@ -86,15 +86,15 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep peer.setFetchAble(true); return; } - } - peer.setFetchAble(true); - if ((chainInventoryMessage.getRemainNum() == 0 && !peer.getSyncBlockToFetch().isEmpty()) - || (chainInventoryMessage.getRemainNum() != 0 - && peer.getSyncBlockToFetch().size() > syncFetchBatchNum)) { - syncService.setFetchFlag(true); - } else { - syncService.syncNext(peer); + peer.setFetchAble(true); + if ((chainInventoryMessage.getRemainNum() == 0 && !peer.getSyncBlockToFetch().isEmpty()) + || (chainInventoryMessage.getRemainNum() != 0 + && peer.getSyncBlockToFetch().size() > syncFetchBatchNum)) { + syncService.setFetchFlag(true); + } else { + syncService.syncNext(peer); + } } } diff --git a/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java b/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java index 7d7457cf2fc..f4b1723a2b9 100644 --- a/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java +++ b/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java @@ -10,7 +10,6 @@ import java.net.InetAddress; import java.net.InetSocketAddress; import java.util.Deque; -import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; @@ -158,7 +157,7 @@ public class PeerConnection { private volatile Pair, Long> syncChainRequested = null; @Setter @Getter - private Set syncBlockInProcess = new HashSet<>(); + private Set syncBlockInProcess = ConcurrentHashMap.newKeySet(); @Setter @Getter private volatile boolean needSyncFromPeer = true; diff --git a/framework/src/main/java/org/tron/core/net/peer/PeerManager.java b/framework/src/main/java/org/tron/core/net/peer/PeerManager.java index f564b90f3ed..5e0e9f61d48 100644 --- a/framework/src/main/java/org/tron/core/net/peer/PeerManager.java +++ b/framework/src/main/java/org/tron/core/net/peer/PeerManager.java @@ -84,13 +84,16 @@ public static synchronized PeerConnection remove(Channel channel) { return peerConnection; } - private static void remove(PeerConnection peerConnection) { - peers.remove(peerConnection); + private static synchronized boolean remove(PeerConnection peerConnection) { + if (!peers.remove(peerConnection)) { + return false; + } if (peerConnection.getChannel().isActive()) { activePeersCount.decrementAndGet(); } else { passivePeersCount.decrementAndGet(); } + return true; } public static synchronized void sortPeers() { @@ -126,13 +129,9 @@ private static void check() { long disconnectTime = peer.getChannel().getDisconnectTime(); if (disconnectTime != 0 && now - disconnectTime > DISCONNECTION_TIME_OUT) { logger.warn("Notify disconnect peer {}.", peer.getInetSocketAddress()); - peers.remove(peer); - if (peer.getChannel().isActive()) { - activePeersCount.decrementAndGet(); - } else { - passivePeersCount.decrementAndGet(); + if (remove(peer)) { + peer.onDisconnect(); } - peer.onDisconnect(); } } } diff --git a/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java b/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java index bda2646abbc..f940e1a0176 100644 --- a/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java +++ b/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java @@ -34,7 +34,7 @@ public class FetchBlockService { @Autowired private ChainBaseManager chainBaseManager; - private FetchBlockInfo fetchBlockInfo = null; + private volatile FetchBlockInfo fetchBlockInfo = null; private final long fetchTimeOut = CommonParameter.getInstance().fetchBlockTimeout; @@ -159,4 +159,4 @@ public FetchBlockInfo(Sha256Hash hash, PeerConnection peer, long time) { } -} \ No newline at end of file +} diff --git a/framework/src/main/java/org/tron/core/net/service/statistics/MessageCount.java b/framework/src/main/java/org/tron/core/net/service/statistics/MessageCount.java index 750f4c52286..edc52fe4d0a 100644 --- a/framework/src/main/java/org/tron/core/net/service/statistics/MessageCount.java +++ b/framework/src/main/java/org/tron/core/net/service/statistics/MessageCount.java @@ -15,7 +15,7 @@ public class MessageCount { private long totalCount = 0; - private void update() { + private synchronized void update() { long time = System.currentTimeMillis() / 1000; long gap = time - indexTime; int k = gap > SIZE ? SIZE : (int) gap; @@ -28,19 +28,19 @@ private void update() { } } - public void add() { + public synchronized void add() { update(); szCount[index]++; totalCount++; } - public void add(int count) { + public synchronized void add(int count) { update(); szCount[index] += count; totalCount += count; } - public int getCount(int interval) { + public synchronized int getCount(int interval) { if (interval > SIZE) { logger.warn("Param interval({}) is gt SIZE({})", interval, SIZE); return 0; @@ -53,16 +53,16 @@ public int getCount(int interval) { return count; } - public long getTotalCount() { + public synchronized long getTotalCount() { return totalCount; } - public void reset() { + public synchronized void reset() { totalCount = 0; } @Override - public String toString() { + public synchronized String toString() { return String.valueOf(totalCount); } diff --git a/framework/src/main/java/org/tron/core/services/WitnessProductBlockService.java b/framework/src/main/java/org/tron/core/services/WitnessProductBlockService.java index e895923114f..ffd73d2835d 100644 --- a/framework/src/main/java/org/tron/core/services/WitnessProductBlockService.java +++ b/framework/src/main/java/org/tron/core/services/WitnessProductBlockService.java @@ -3,10 +3,10 @@ import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import java.util.Arrays; -import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; @@ -20,7 +20,7 @@ public class WitnessProductBlockService { private Cache historyBlockCapsuleCache = CacheBuilder.newBuilder() .initialCapacity(200).maximumSize(200).build(); - private Map cheatWitnessInfoMap = new HashMap<>(); + private Map cheatWitnessInfoMap = new ConcurrentHashMap<>(); public void validWitnessProductTwoBlock(BlockCapsule block) { try { diff --git a/framework/src/test/java/org/tron/common/backup/BackupManagerTest.java b/framework/src/test/java/org/tron/common/backup/BackupManagerTest.java index 5ff02fc8cb5..78060ea0df1 100644 --- a/framework/src/test/java/org/tron/common/backup/BackupManagerTest.java +++ b/framework/src/test/java/org/tron/common/backup/BackupManagerTest.java @@ -2,6 +2,7 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.lang.reflect.Modifier; import java.net.InetAddress; import java.net.InetSocketAddress; import java.util.ArrayList; @@ -52,6 +53,13 @@ public void tearDown() { Args.clearParam(); } + @Test + public void statusIsVolatileForCrossThreadVisibility() throws Exception { + Field status = BackupManager.class.getDeclaredField("status"); + + Assert.assertTrue(Modifier.isVolatile(status.getModifiers())); + } + @Test public void test() throws Exception { CommonParameter.getInstance().setBackupPriority(8); diff --git a/framework/src/test/java/org/tron/core/metrics/blockchain/BlockChainMetricManagerTest.java b/framework/src/test/java/org/tron/core/metrics/blockchain/BlockChainMetricManagerTest.java new file mode 100644 index 00000000000..71269d0c7f7 --- /dev/null +++ b/framework/src/test/java/org/tron/core/metrics/blockchain/BlockChainMetricManagerTest.java @@ -0,0 +1,38 @@ +package org.tron.core.metrics.blockchain; + +import java.lang.reflect.Method; +import java.util.List; +import org.junit.Assert; +import org.junit.Test; +import org.tron.common.parameter.CommonParameter; +import org.tron.core.metrics.MetricsKey; +import org.tron.core.metrics.MetricsUtil; + +public class BlockChainMetricManagerTest { + + @Test + @SuppressWarnings("unchecked") + public void missingDuplicateWitnessBlockNumberDefaultsToZero() throws Exception { + CommonParameter parameter = CommonParameter.getInstance(); + boolean nodeMetricsEnabled = parameter.isNodeMetricsEnable(); + String witness = "missing-block-number-" + System.nanoTime(); + parameter.setNodeMetricsEnable(true); + try { + MetricsUtil.counterInc(MetricsKey.BLOCKCHAIN_DUP_WITNESS + witness); + + Method getDupWitness = BlockChainMetricManager.class.getDeclaredMethod("getDupWitness"); + getDupWitness.setAccessible(true); + List dupWitnesses = (List) getDupWitness.invoke( + new BlockChainMetricManager()); + + DupWitnessInfo dupWitness = dupWitnesses.stream() + .filter(info -> witness.equals(info.getAddress())) + .findFirst() + .orElse(null); + Assert.assertNotNull(dupWitness); + Assert.assertEquals(0L, dupWitness.getBlockNum()); + } finally { + parameter.setNodeMetricsEnable(nodeMetricsEnabled); + } + } +} diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java index 56853c3dbb7..08eaa1e2abc 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java @@ -1,21 +1,28 @@ package org.tron.core.net.messagehandler; import java.util.ArrayList; +import java.util.Arrays; import java.util.LinkedList; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.AfterClass; import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; +import org.mockito.Mockito; import org.tron.common.TestConstants; import org.tron.common.utils.Pair; +import org.tron.common.utils.ReflectUtils; +import org.tron.common.utils.Sha256Hash; import org.tron.core.capsule.BlockCapsule.BlockId; import org.tron.core.config.Parameter.NetConstants; import org.tron.core.config.args.Args; import org.tron.core.exception.P2pException; +import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.keepalive.PingMessage; import org.tron.core.net.message.sync.ChainInventoryMessage; import org.tron.core.net.peer.PeerConnection; +import org.tron.core.net.service.sync.SyncService; public class ChainInventoryMsgHandlerTest { @@ -78,4 +85,74 @@ public void testProcessMessage() throws Exception { Assert.assertNull(msg.getAnswerMessage()); } + @Test + public void testFetchFlagDecisionIsMadeUnderBlockLock() throws Exception { + Object blockLock = new Object(); + TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class); + SyncService syncService = Mockito.mock(SyncService.class); + PeerConnection testPeer = Mockito.spy(new PeerConnection()); + AtomicBoolean fetchAbleSetUnderLock = new AtomicBoolean(); + AtomicBoolean fetchFlagSetUnderLock = new AtomicBoolean(); + BlockId firstBlock = new BlockId(Sha256Hash.ZERO_HASH, 1); + BlockId secondBlock = new BlockId(Sha256Hash.ZERO_HASH, 2); + + Mockito.when(tronNetDelegate.getBlockLock()).thenReturn(blockLock); + Mockito.when(tronNetDelegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, 0)); + Mockito.when(tronNetDelegate.containBlock(Mockito.any())).thenReturn(false); + Mockito.doAnswer(invocation -> { + if (invocation.getArgument(0)) { + fetchAbleSetUnderLock.set(Thread.holdsLock(blockLock)); + } + return invocation.callRealMethod(); + }).when(testPeer).setFetchAble(Mockito.anyBoolean()); + Mockito.doAnswer(invocation -> { + fetchFlagSetUnderLock.set(Thread.holdsLock(blockLock)); + return null; + }).when(syncService).setFetchFlag(true); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", tronNetDelegate); + ReflectUtils.setFieldValue(handler, "syncService", syncService); + testPeer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(firstBlock)), + System.currentTimeMillis())); + + handler.processMessage(testPeer, + new ChainInventoryMessage(Arrays.asList(firstBlock, secondBlock), 0L)); + + Assert.assertTrue(fetchAbleSetUnderLock.get()); + Assert.assertTrue(fetchFlagSetUnderLock.get()); + } + + @Test + public void testSyncNextDecisionIsMadeUnderBlockLock() throws Exception { + Object blockLock = new Object(); + TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class); + SyncService syncService = Mockito.mock(SyncService.class); + PeerConnection testPeer = Mockito.spy(new PeerConnection()); + AtomicBoolean fetchAbleSetUnderLock = new AtomicBoolean(); + AtomicBoolean syncNextCalledUnderLock = new AtomicBoolean(); + BlockId firstBlock = new BlockId(Sha256Hash.ZERO_HASH, 1); + + Mockito.when(tronNetDelegate.getBlockLock()).thenReturn(blockLock); + Mockito.when(tronNetDelegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, 0)); + Mockito.when(tronNetDelegate.containBlock(Mockito.any())).thenReturn(false); + Mockito.doAnswer(invocation -> { + if (invocation.getArgument(0)) { + fetchAbleSetUnderLock.set(Thread.holdsLock(blockLock)); + } + return invocation.callRealMethod(); + }).when(testPeer).setFetchAble(Mockito.anyBoolean()); + Mockito.doAnswer(invocation -> { + syncNextCalledUnderLock.set(Thread.holdsLock(blockLock)); + return null; + }).when(syncService).syncNext(testPeer); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", tronNetDelegate); + ReflectUtils.setFieldValue(handler, "syncService", syncService); + testPeer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(firstBlock)), + System.currentTimeMillis())); + + handler.processMessage(testPeer, new ChainInventoryMessage(Arrays.asList(firstBlock), 0L)); + + Assert.assertTrue(fetchAbleSetUnderLock.get()); + Assert.assertTrue(syncNextCalledUnderLock.get()); + } + } diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java index cc30fb70b0b..406e3358c5f 100644 --- a/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java +++ b/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java @@ -6,6 +6,7 @@ import java.util.ArrayList; import java.util.LinkedList; import java.util.List; +import java.util.concurrent.ConcurrentHashMap; import org.junit.AfterClass; import org.junit.Assert; @@ -56,6 +57,14 @@ public void testVariableDefaultValue() { Assert.assertTrue(!peerConnection.isSyncFinish()); } + @Test + public void testSyncBlockInProcessUsesConcurrentSet() { + PeerConnection peerConnection = new PeerConnection(); + + Assert.assertTrue(peerConnection.getSyncBlockInProcess() + instanceof ConcurrentHashMap.KeySetView); + } + @Test public void testOnDisconnect() { PeerConnection peerConnection = new PeerConnection(); diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerManagerTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerManagerTest.java index ffba127a6fd..bc0f112c5d5 100644 --- a/framework/src/test/java/org/tron/core/net/peer/PeerManagerTest.java +++ b/framework/src/test/java/org/tron/core/net/peer/PeerManagerTest.java @@ -3,10 +3,12 @@ import static org.mockito.Mockito.mock; import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import org.junit.After; import org.junit.AfterClass; @@ -188,4 +190,50 @@ public void testSortPeers() throws Exception { Assert.assertEquals(PeerManager.getPeers().get(0), p2); } + @Test + public void checkDoesNotRemoveOrCleanUpPeerAfterDisconnectCallback() throws Exception { + assertCheckDoesNotRemovePeerTwice(true); + assertCheckDoesNotRemovePeerTwice(false); + } + + private void assertCheckDoesNotRemovePeerTwice(boolean active) throws Exception { + AtomicReference removedByCallback = new AtomicReference<>(); + Channel channel = new Channel() { + @Override + public boolean isActive() { + return active; + } + + @Override + public long getDisconnectTime() { + removedByCallback.set(PeerManager.remove(this)); + return System.currentTimeMillis() - 120_000; + } + }; + PeerConnection peer = mock(PeerConnection.class); + Mockito.when(peer.getChannel()).thenReturn(channel); + + Field peersField = PeerManager.class.getDeclaredField("peers"); + peersField.setAccessible(true); + peersField.set(null, Collections.synchronizedList( + new ArrayList<>(Collections.singletonList(peer)))); + PeerManager.getActivePeersCount().set(active ? 1 : 0); + PeerManager.getPassivePeersCount().set(active ? 0 : 1); + + Method check = PeerManager.class.getDeclaredMethod("check"); + check.setAccessible(true); + try { + check.invoke(null); + + Assert.assertSame(peer, removedByCallback.get()); + Assert.assertEquals(0, PeerManager.getActivePeersCount().get()); + Assert.assertEquals(0, PeerManager.getPassivePeersCount().get()); + Mockito.verify(peer, Mockito.never()).onDisconnect(); + } finally { + peersField.set(null, Collections.synchronizedList(new ArrayList<>())); + PeerManager.getActivePeersCount().set(0); + PeerManager.getPassivePeersCount().set(0); + } + } + } diff --git a/framework/src/test/java/org/tron/core/net/service/fetchblock/FetchBlockServiceTest.java b/framework/src/test/java/org/tron/core/net/service/fetchblock/FetchBlockServiceTest.java new file mode 100644 index 00000000000..a5eb2751be0 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/service/fetchblock/FetchBlockServiceTest.java @@ -0,0 +1,16 @@ +package org.tron.core.net.service.fetchblock; + +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import org.junit.Assert; +import org.junit.Test; + +public class FetchBlockServiceTest { + + @Test + public void testFetchBlockInfoIsVolatile() throws Exception { + Field fetchBlockInfo = FetchBlockService.class.getDeclaredField("fetchBlockInfo"); + + Assert.assertTrue(Modifier.isVolatile(fetchBlockInfo.getModifiers())); + } +} diff --git a/framework/src/test/java/org/tron/core/net/service/statistics/MessageCountTest.java b/framework/src/test/java/org/tron/core/net/service/statistics/MessageCountTest.java new file mode 100644 index 00000000000..8bdd4f4d8cb --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/service/statistics/MessageCountTest.java @@ -0,0 +1,114 @@ +package org.tron.core.net.service.statistics; + +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.Assert; +import org.junit.Test; + +public class MessageCountTest { + + private static final int THREAD_COUNT = 8; + private static final int ITERATIONS = 50_000; + + @Test(timeout = 10_000) + public void windowOperationsUseOneMonitor() throws Exception { + MessageCount messageCount = new MessageCount(); + Method update = MessageCount.class.getDeclaredMethod("update"); + update.setAccessible(true); + List operations = Arrays.asList( + messageCount::add, + () -> messageCount.add(2), + () -> messageCount.getCount(1), + messageCount::getTotalCount, + messageCount::reset, + messageCount::toString, + () -> update.invoke(messageCount)); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + for (CheckedRunnable operation : operations) { + assertUsesMonitor(messageCount, executor, operation); + } + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS)); + } + } + + @Test(timeout = 10_000) + public void concurrentAddsDoNotLoseUpdates() throws Exception { + MessageCount messageCount = new MessageCount(); + ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT); + CountDownLatch ready = new CountDownLatch(THREAD_COUNT); + CountDownLatch start = new CountDownLatch(1); + try { + Future[] futures = new Future[THREAD_COUNT]; + for (int thread = 0; thread < THREAD_COUNT; thread++) { + futures[thread] = executor.submit(() -> { + ready.countDown(); + try { + start.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + for (int i = 0; i < ITERATIONS; i++) { + if ((i & 1) == 0) { + messageCount.add(); + } else { + messageCount.add(2); + } + } + }); + } + + Assert.assertTrue(ready.await(1, TimeUnit.SECONDS)); + start.countDown(); + for (Future future : futures) { + future.get(5, TimeUnit.SECONDS); + } + + int expected = THREAD_COUNT * ITERATIONS / 2 * 3; + Assert.assertEquals(expected, messageCount.getTotalCount()); + Assert.assertEquals(expected, messageCount.getCount(60)); + } finally { + start.countDown(); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS)); + } + } + + private void assertUsesMonitor(MessageCount messageCount, ExecutorService executor, + CheckedRunnable operation) throws Exception { + CountDownLatch started = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(1); + Future future; + synchronized (messageCount) { + future = executor.submit(() -> { + started.countDown(); + try { + operation.run(); + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + finished.countDown(); + } + }); + Assert.assertTrue(started.await(1, TimeUnit.SECONDS)); + Assert.assertFalse(finished.await(100, TimeUnit.MILLISECONDS)); + } + future.get(1, TimeUnit.SECONDS); + } + + @FunctionalInterface + private interface CheckedRunnable { + + void run() throws Exception; + } +} diff --git a/framework/src/test/java/org/tron/core/services/WitnessProductBlockServiceTest.java b/framework/src/test/java/org/tron/core/services/WitnessProductBlockServiceTest.java index 0b915b766b9..6f999616b4a 100644 --- a/framework/src/test/java/org/tron/core/services/WitnessProductBlockServiceTest.java +++ b/framework/src/test/java/org/tron/core/services/WitnessProductBlockServiceTest.java @@ -4,6 +4,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Assert; import org.junit.Test; @@ -13,6 +14,14 @@ public class WitnessProductBlockServiceTest { + @Test + public void cheatWitnessInfoMapSupportsConcurrentUpdates() { + WitnessProductBlockService witnessProductBlockService = new WitnessProductBlockService(); + + Assert.assertTrue(witnessProductBlockService.queryCheatWitnessInfo() + instanceof ConcurrentHashMap); + } + @Test public void GetSetCheatWitnessInfoTest() { WitnessProductBlockService.CheatWitnessInfo cheatWitnessInfo =