diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java b/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java index f399ab2f3b4..0321506b504 100644 --- a/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java +++ b/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java @@ -82,7 +82,6 @@ public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting private static final String MSG_ID_PREFIX = "msgId:"; private static final String FUTURES_PREFIX = "futures:"; private static final String SINGLE_LOG_POSTFIX = ";"; - private static final int MAX_MERGE_SEND_MILLS = 1; private static final String THREAD_PREFIX_SPLIT_CHAR = "_"; private static final int MAX_MERGE_SEND_THREAD = 1; private static final long KEEP_ALIVE_TIME = Integer.MAX_VALUE; @@ -587,7 +586,15 @@ public void run() { while (true) { mergeLock.lock(); try { - mergeCondition.await(MAX_MERGE_SEND_MILLS, TimeUnit.MILLISECONDS); + // Park until there are pending messages, so the merge thread no longer + // burns CPU with a 1ms polling cycle when idle. The check-and-wait is + // atomic under mergeLock and producers offer to the basket before + // signalling (see sendSyncRequest), so no wake-up can be lost. + while (isBasketEmpty()) { + isSending = false; + mergeCondition.await(); + } + isSending = true; } catch (InterruptedException e) { Thread.currentThread().interrupt(); LOGGER.warn("MergedSendRunnable wait interrupted", e); @@ -639,6 +646,21 @@ public void run() { } } + /** + * Checks whether all baskets are empty. The merge thread parks itself + * when this returns true, avoiding the idle 1ms polling busy loop. + * + * @return true if every basket in basketMap is empty + */ + private boolean isBasketEmpty() { + for (BlockingQueue basket : basketMap.values()) { + if (!basket.isEmpty()) { + return false; + } + } + return true; + } + private void printMergeMessageLog(MergedWarpMessage mergeMessage) { if (LOGGER.isDebugEnabled()) { LOGGER.debug("merge msg size:{}", mergeMessage.msgIds.size()); diff --git a/core/src/test/java/org/apache/seata/core/rpc/netty/NettyRemotingClientBehaviorTest.java b/core/src/test/java/org/apache/seata/core/rpc/netty/NettyRemotingClientBehaviorTest.java index 215ead96b82..0a210954796 100644 --- a/core/src/test/java/org/apache/seata/core/rpc/netty/NettyRemotingClientBehaviorTest.java +++ b/core/src/test/java/org/apache/seata/core/rpc/netty/NettyRemotingClientBehaviorTest.java @@ -38,6 +38,7 @@ import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; @@ -1479,6 +1480,52 @@ public void testMergedSendRunnableWithEmptyBasket() throws Exception { } } + @Test + public void testMergedSendRunnableIdleWaitState() throws Exception { + TestNettyRemotingClientWithMergeRunnable mergeClient = + new TestNettyRemotingClientWithMergeRunnable(clientConfig, messageExecutor); + + try { + mergeClient.init(); + + // Wait for the merge send thread to start and park itself + Thread.sleep(300); + + // The merge thread must be parked on Condition.await (WAITING) instead of + // spinning on a 1ms timed wait (TIMED_WAITING) when idle + for (Map.Entry entry : + Thread.getAllStackTraces().entrySet()) { + Thread thread = entry.getKey(); + if (thread.getName().startsWith("rpcMergeMessageSend")) { + assertFalse( + Thread.State.TIMED_WAITING == thread.getState(), + "merge send thread should not spin on a 1ms timed wait when idle: " + thread.getName()); + } + } + + // A message must wake the thread and get drained from the basket + GlobalBeginRequest request = new GlobalBeginRequest(); + request.setTransactionName("test-tx-idle-wake"); + try { + mergeClient.sendSyncRequest(request); + } catch (Exception e) { + // Expected: no real server at 127.0.0.1:8080, the merge thread + // drains the basket and fast-fails the future + } + + Thread.sleep(300); + assertTrue( + mergeClient.basketMap.values().stream().allMatch(BlockingQueue::isEmpty), + "basket should be drained by the merge send thread after wake-up"); + } finally { + try { + mergeClient.destroy(); + } catch (Exception e) { + // Ignore + } + } + } + /** * Test implementation that simulates reconnect exception */