diff --git a/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java b/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java index 7d97e6f4ba9..7337f3df175 100644 --- a/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java +++ b/framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java @@ -27,22 +27,21 @@ public static NativeMessageQueue getInstance() { } public boolean start(int bindPort, int sendQueueLength) { - context = new ZContext(); - publisher = context.createSocket(SocketType.PUB); - - if (Objects.isNull(publisher)) { - return false; - } - - if (bindPort == 0 || bindPort < 0) { + if (bindPort <= 0) { bindPort = DEFAULT_BIND_PORT; } - if (sendQueueLength < 0) { + if (sendQueueLength <= 0) { sendQueueLength = DEFAULT_QUEUE_LENGTH; } + context = new ZContext(); context.setSndHWM(sendQueueLength); + publisher = context.createSocket(SocketType.PUB); + + if (Objects.isNull(publisher)) { + return false; + } String bindAddress = String.format("tcp://*:%d", bindPort); return publisher.bind(bindAddress); diff --git a/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java b/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java index 7efc10a8ed5..65c2adf573d 100644 --- a/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java +++ b/framework/src/main/java/org/tron/core/services/event/BlockEventLoad.java @@ -37,7 +37,7 @@ public class BlockEventLoad { public void init() { executor.scheduleWithFixedDelay(() -> { try { - if (!instance.isBusy()) { + if (!instance.isBusy() && !realtimeEventService.isBusy()) { load(); } } catch (Exception e) { diff --git a/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java b/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java index cef16cd81c1..d2ec6a83c36 100644 --- a/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java +++ b/framework/src/main/java/org/tron/core/services/event/RealtimeEventService.java @@ -29,7 +29,7 @@ public class RealtimeEventService { private static BlockingQueue queue = new LinkedBlockingQueue<>(); - private int maxEventSize = 10000; + private static final int BUSY_EVENT_SIZE = 500; private final ScheduledExecutorService executor = ExecutorServiceManager .newSingleThreadScheduledExecutor("realtime-event"); @@ -56,13 +56,13 @@ public void close() { } public void add(Event event) { - if (queue.size() >= maxEventSize) { - logger.warn("Add event failed, blockId {}.", event.getBlockEvent().getBlockId().getString()); - return; - } queue.offer(event); } + public boolean isBusy() { + return queue.size() >= BUSY_EVENT_SIZE; + } + public synchronized void work() { while (queue.size() > 0) { Event event = queue.poll(); diff --git a/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java b/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java index 5219654977b..1e7aeefbd09 100644 --- a/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java +++ b/framework/src/test/java/org/tron/common/logsfilter/NativeMessageQueueTest.java @@ -1,5 +1,6 @@ package org.tron.common.logsfilter; +import java.lang.reflect.Field; import java.util.concurrent.ExecutorService; import org.junit.After; import org.junit.Assert; @@ -21,10 +22,20 @@ public class NativeMessageQueueTest { @After public void tearDown() { + NativeMessageQueue.getInstance().stop(); ExecutorServiceManager.shutdownAndAwaitTermination(subscriberExecutor, zmqSubscriber); subscriberExecutor = null; } + @Test + public void configuredSendQueueLengthIsAppliedToPublisherSocket() throws Exception { + int sendQueueLength = 2000; + + Assert.assertTrue(NativeMessageQueue.getInstance().start(bindPort, sendQueueLength)); + + Assert.assertEquals(sendQueueLength, getPublisher().getSndHWM()); + } + @Test public void invalidBindPort() { boolean bRet = NativeMessageQueue.getInstance().start(-1111, 0); @@ -33,10 +44,17 @@ public void invalidBindPort() { } @Test - public void invalidSendLength() { - boolean bRet = NativeMessageQueue.getInstance().start(0, -2222); - Assert.assertEquals(true, bRet); - NativeMessageQueue.getInstance().stop(); + public void negativeSendQueueLengthUsesDefaultSndHWM() throws Exception { + Assert.assertTrue(NativeMessageQueue.getInstance().start(bindPort, -1)); + + Assert.assertEquals(1000, getPublisher().getSndHWM()); + } + + @Test + public void zeroSendQueueLengthUsesDefaultSndHWM() throws Exception { + Assert.assertTrue(NativeMessageQueue.getInstance().start(bindPort, 0)); + + Assert.assertEquals(1000, getPublisher().getSndHWM()); } @Test @@ -84,4 +102,10 @@ public void startSubscribeThread() { } }); } + + private ZMQ.Socket getPublisher() throws ReflectiveOperationException { + Field publisherField = NativeMessageQueue.class.getDeclaredField("publisher"); + publisherField.setAccessible(true); + return (ZMQ.Socket) publisherField.get(NativeMessageQueue.getInstance()); + } } diff --git a/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java b/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java index 991133fee78..01f52ed2424 100644 --- a/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java +++ b/framework/src/test/java/org/tron/core/event/BlockEventLoadTest.java @@ -5,9 +5,11 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ScheduledExecutorService; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; +import org.tron.common.logsfilter.EventPluginLoader; import org.tron.common.utils.ReflectUtils; import org.tron.core.ChainBaseManager; import org.tron.core.capsule.BlockCapsule; @@ -23,6 +25,31 @@ public class BlockEventLoadTest { BlockEventLoad blockEventLoad = new BlockEventLoad(); + @Test(timeout = 2_000) + public void shouldNotLoadWhenRealtimeEventServiceIsBusy() throws Exception { + EventPluginLoader eventPluginLoader = mock(EventPluginLoader.class); + RealtimeEventService realtimeEventService = mock(RealtimeEventService.class); + Manager manager = mock(Manager.class); + ReflectUtils.setFieldValue(blockEventLoad, "instance", eventPluginLoader); + ReflectUtils.setFieldValue(blockEventLoad, "realtimeEventService", realtimeEventService); + ReflectUtils.setFieldValue(blockEventLoad, "manager", manager); + Mockito.when(eventPluginLoader.isBusy()).thenReturn(false); + Mockito.when(realtimeEventService.isBusy()).thenReturn(true); + + Field executorField = BlockEventLoad.class.getDeclaredField("executor"); + executorField.setAccessible(true); + ScheduledExecutorService executor = (ScheduledExecutorService) executorField + .get(blockEventLoad); + try { + blockEventLoad.init(); + + Mockito.verify(realtimeEventService, Mockito.timeout(1_000).atLeastOnce()).isBusy(); + Mockito.verifyNoInteractions(manager); + } finally { + executor.shutdownNow(); + } + } + @Test public void test() throws Exception { Method method = blockEventLoad.getClass().getDeclaredMethod("load"); diff --git a/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java b/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java index f58f725195c..4d6cb71bf85 100644 --- a/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java +++ b/framework/src/test/java/org/tron/core/event/RealtimeEventServiceTest.java @@ -3,8 +3,10 @@ import static org.mockito.Mockito.mock; import com.google.protobuf.ByteString; +import java.lang.reflect.Field; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.BlockingQueue; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; @@ -26,6 +28,33 @@ public class RealtimeEventServiceTest { RealtimeEventService realtimeEventService = new RealtimeEventService(); + @Test + public void shouldBecomeBusyAt500EventsAndRetainLaterEvents() throws Exception { + Field queueField = RealtimeEventService.class.getDeclaredField("queue"); + queueField.setAccessible(true); + BlockingQueue queue = (BlockingQueue) queueField.get(null); + queue.clear(); + + try { + Event event = mock(Event.class); + for (int i = 0; i < 499; i++) { + realtimeEventService.add(event); + } + + Assert.assertFalse(realtimeEventService.isBusy()); + + realtimeEventService.add(event); + Assert.assertTrue(realtimeEventService.isBusy()); + + realtimeEventService.add(event); + Assert.assertEquals(501, queue.size()); + } finally { + queue.clear(); + } + + Assert.assertFalse(realtimeEventService.isBusy()); + } + @Test public void test() throws Exception { BlockEvent be1 = new BlockEvent();