diff --git a/framework/src/main/java/org/tron/common/backup/socket/BackupServer.java b/framework/src/main/java/org/tron/common/backup/socket/BackupServer.java index 67739ac50d2..7ac1b4db296 100644 --- a/framework/src/main/java/org/tron/common/backup/socket/BackupServer.java +++ b/framework/src/main/java/org/tron/common/backup/socket/BackupServer.java @@ -27,12 +27,12 @@ public class BackupServer implements AutoCloseable { private BackupManager backupManager; - private Channel channel; - private volatile boolean shutdown = false; private final String name = "BackupServer"; - private ExecutorService executor; + private volatile ExecutorService executor; + + private volatile NioEventLoopGroup group; @Autowired public BackupServer(final BackupManager backupManager) { @@ -41,6 +41,14 @@ public BackupServer(final BackupManager backupManager) { public void initServer() { if (port > 0 && commonParameter.getBackupMembers().size() > 0) { + try { + // Let close() reach the group before bind() completes. + group = new NioEventLoopGroup(1); + } catch (RuntimeException e) { + // Do not propagate selector failures to block processing. + logger.error("Start backup server with port {} failed.", port, e); + return; + } executor = ExecutorServiceManager.newSingleThreadExecutor(name); executor.submit(() -> { try { @@ -53,7 +61,6 @@ public void initServer() { } private void start() throws Exception { - NioEventLoopGroup group = new NioEventLoopGroup(1); try { while (!shutdown) { Bootstrap b = new Bootstrap(); @@ -73,7 +80,7 @@ public void initChannel(NioDatagramChannel ch) } }); - channel = b.bind(port).sync().channel(); + Channel channel = b.bind(port).sync().channel(); logger.info("Backup server started, bind port {}", port); @@ -84,10 +91,15 @@ public void initChannel(NioDatagramChannel ch) } logger.warn("Restart backup server ..."); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (!shutdown) { + logger.error("Backup server interrupted.", e); + } } catch (Exception e) { logger.error("Start backup server with port {} failed.", port, e); } finally { - group.shutdownGracefully().sync(); + group.shutdownGracefully(); } } @@ -96,14 +108,12 @@ public void close() { logger.info("Closing backup server..."); shutdown = true; backupManager.stop(); - if (channel != null) { - try { - channel.close().await(10, TimeUnit.SECONDS); - } catch (Exception e) { - logger.warn("Closing backup server failed.", e); - } + if (executor != null) { + executor.shutdownNow(); + } + if (group != null) { + group.shutdownGracefully().awaitUninterruptibly(10, TimeUnit.SECONDS); } - ExecutorServiceManager.shutdownAndAwaitTermination(executor, name); logger.info("Backup server closed."); } } 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..540ca996635 100644 --- a/framework/src/test/java/org/tron/common/backup/BackupManagerTest.java +++ b/framework/src/test/java/org/tron/common/backup/BackupManagerTest.java @@ -9,7 +9,6 @@ import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; import java.util.function.BiFunction; import org.junit.After; @@ -138,15 +137,8 @@ public void testSendKeepAliveMessage() throws Exception { Thread.sleep(parameter.getKeepAliveInterval() + 1000);//test send KeepAliveMessage - field = manager.getClass().getDeclaredField("executorService"); - field.setAccessible(true); - ScheduledExecutorService executorService = (ScheduledExecutorService) field.get(manager); - executorService.shutdown(); - - Field field2 = backupServer.getClass().getDeclaredField("executor"); - field2.setAccessible(true); - ExecutorService executorService2 = (ExecutorService) field2.get(backupServer); - executorService2.shutdown(); + // also stops the manager's keep-alive executor + backupServer.close(); Assert.assertEquals(BackupManager.BackupStatusEnum.INIT, manager.getStatus()); } diff --git a/framework/src/test/java/org/tron/common/backup/BackupServerTest.java b/framework/src/test/java/org/tron/common/backup/BackupServerTest.java index 50778970d87..d892452a681 100644 --- a/framework/src/test/java/org/tron/common/backup/BackupServerTest.java +++ b/framework/src/test/java/org/tron/common/backup/BackupServerTest.java @@ -1,17 +1,34 @@ package org.tron.common.backup; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mockStatic; + +import io.netty.channel.Channel; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.concurrent.SingleThreadEventExecutor; +import java.io.IOException; +import java.net.DatagramSocket; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; import org.junit.After; +import org.junit.Assert; import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.junit.rules.Timeout; +import org.mockito.MockedStatic; +import org.springframework.test.util.ReflectionTestUtils; import org.tron.common.TestConstants; import org.tron.common.backup.socket.BackupServer; +import org.tron.common.es.ExecutorServiceManager; import org.tron.common.parameter.CommonParameter; -import org.tron.common.utils.PublicMethod; import org.tron.core.config.args.Args; @@ -22,17 +39,18 @@ public class BackupServerTest { @Rule public Timeout globalTimeout = Timeout.seconds(60); + private BackupManager backupManager; private BackupServer backupServer; @Before public void setUp() throws Exception { Args.setParam(new String[]{"-d", temporaryFolder.newFolder().toString()}, TestConstants.TEST_CONF); - CommonParameter.getInstance().setBackupPort(PublicMethod.chooseRandomPort()); + CommonParameter.getInstance().setBackupPort(freeUdpPort()); List members = new ArrayList<>(); members.add("127.0.0.2"); CommonParameter.getInstance().setBackupMembers(members); - BackupManager backupManager = new BackupManager(); + backupManager = new BackupManager(); backupManager.init(); backupServer = new BackupServer(backupManager); } @@ -43,10 +61,104 @@ public void tearDown() { Args.clearParam(); } - @Test(timeout = 60_000) - public void test() throws InterruptedException { + @Test + public void closeAfterStarted() throws Exception { backupServer.initServer(); - // wait for the server to start so channel is assigned before close() is called - Thread.sleep(1000); + assertTrue("server did not bind in time", waitUntil(() -> { + Channel channel = serverChannel(); + return channel != null && channel.isActive(); + }, 30)); + + Channel channel = serverChannel(); + ExecutorService executor = + (ExecutorService) ReflectionTestUtils.getField(backupServer, "executor"); + backupServer.close(); + + Assert.assertNotNull(channel); + assertFalse("server channel is still open", channel.isOpen()); + assertTrue("event loop group did not terminate", eventLoopGroup().isTerminated()); + Assert.assertNotNull(executor); + assertTrue("server executor did not terminate", + executor.awaitTermination(5, TimeUnit.SECONDS)); + } + + @Test + public void closeWhileBindIsPending() throws Exception { + ExecutorService executor = ExecutorServiceManager.newSingleThreadExecutor("BackupServer"); + CountDownLatch parked = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + FutureTask closed = new FutureTask<>(backupServer::close, null); + Thread closer = new Thread(closed, "backup-test-closer"); + try { + try (MockedStatic factory = + mockStatic(ExecutorServiceManager.class)) { + factory.when(() -> ExecutorServiceManager.newSingleThreadExecutor("BackupServer")) + .thenAnswer(invocation -> { + // The group exists, but the server task has not been submitted yet. + eventLoopGroup().next().execute(() -> { + parked.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + assertTrue("event loop did not park", parked.await(5, TimeUnit.SECONDS)); + return executor; + }); + backupServer.initServer(); + } + + NioEventLoopGroup group = eventLoopGroup(); + SingleThreadEventExecutor loop = (SingleThreadEventExecutor) group.next(); + assertTrue("bind registration did not queue", waitUntil(() -> loop.pendingTasks() > 0, 5)); + closer.start(); + assertTrue("close did not shut down the group", waitUntil(group::isShuttingDown, 5)); + // Let bind proceed only after close() has requested shutdown. + release.countDown(); + closed.get(15, TimeUnit.SECONDS); + assertTrue("server executor did not terminate", + executor.awaitTermination(5, TimeUnit.SECONDS)); + assertTrue("event loop group did not terminate", group.isTerminated()); + } finally { + // Release the event loop even when a preparation or shutdown assertion fails. + release.countDown(); + closer.interrupt(); + executor.shutdownNow(); + NioEventLoopGroup group = eventLoopGroup(); + if (group != null) { + assertTrue("event loop cleanup failed", + group.shutdownGracefully().awaitUninterruptibly(10, TimeUnit.SECONDS)); + } + closer.join(10_000); + assertTrue("executor cleanup failed", executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + private Channel serverChannel() { + Object handler = ReflectionTestUtils.getField(backupManager, "messageHandler"); + return handler == null ? null : (Channel) ReflectionTestUtils.getField(handler, "channel"); + } + + private NioEventLoopGroup eventLoopGroup() { + return (NioEventLoopGroup) ReflectionTestUtils.getField(backupServer, "group"); + } + + private boolean waitUntil(BooleanSupplier condition, long timeoutSeconds) + throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds); + while (!condition.getAsBoolean()) { + if (System.nanoTime() - deadline >= 0) { + return false; + } + TimeUnit.MILLISECONDS.sleep(10); + } + return true; + } + + private int freeUdpPort() throws IOException { + try (DatagramSocket socket = new DatagramSocket(0)) { + return socket.getLocalPort(); + } } }