From 19681e495cd56ff637b5e6248079ac1260528828 Mon Sep 17 00:00:00 2001 From: wb Date: Fri, 28 Aug 2026 18:18:21 +0800 Subject: [PATCH] fix(net): harden shared state concurrency Network workers, synchronization tasks, metrics collectors, and backup services access several mutable states across threads without consistent visibility or synchronization guarantees. Use concurrent collections and synchronized state transitions where compound access must remain atomic, publish cross-thread fields safely, serialize peer cleanup with existing manager operations, and prevent duplicate-witness metrics from exposing incomplete companion state. --- .../org/tron/common/backup/BackupManager.java | 2 +- .../blockchain/BlockChainMetricManager.java | 4 +- .../ChainInventoryMsgHandler.java | 16 +-- .../tron/core/net/peer/PeerConnection.java | 3 +- .../org/tron/core/net/peer/PeerManager.java | 15 ++- .../service/fetchblock/FetchBlockService.java | 4 +- .../net/service/statistics/MessageCount.java | 14 +-- .../services/WitnessProductBlockService.java | 4 +- .../tron/common/backup/BackupManagerTest.java | 8 ++ .../BlockChainMetricManagerTest.java | 38 ++++++ .../ChainInventoryMsgHandlerTest.java | 77 ++++++++++++ .../core/net/peer/PeerConnectionTest.java | 9 ++ .../tron/core/net/peer/PeerManagerTest.java | 48 ++++++++ .../fetchblock/FetchBlockServiceTest.java | 16 +++ .../service/statistics/MessageCountTest.java | 114 ++++++++++++++++++ .../WitnessProductBlockServiceTest.java | 9 ++ 16 files changed, 349 insertions(+), 32 deletions(-) create mode 100644 framework/src/test/java/org/tron/core/metrics/blockchain/BlockChainMetricManagerTest.java create mode 100644 framework/src/test/java/org/tron/core/net/service/fetchblock/FetchBlockServiceTest.java create mode 100644 framework/src/test/java/org/tron/core/net/service/statistics/MessageCountTest.java 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 =