From 5485b491bdecf1dac8d79edd1f98ae3f8e996471 Mon Sep 17 00:00:00 2001 From: Evgeny Malygin Date: Fri, 4 Sep 2026 11:46:26 -0400 Subject: [PATCH 1/2] IT[NettyTcpConnectionImplIT]: add is writeable test Signed-off-by: Evgeny Malygin --- .../bmq/it/NettyTcpConnectionImplIT.java | 85 +++++++++++++++++++ .../bmq/it/util/BmqBrokerSimulator.java | 16 ++-- 2 files changed, 96 insertions(+), 5 deletions(-) diff --git a/bmq-sdk/src/test/java/com/bloomberg/bmq/it/NettyTcpConnectionImplIT.java b/bmq-sdk/src/test/java/com/bloomberg/bmq/it/NettyTcpConnectionImplIT.java index cc124e47..f9dcd9b0 100644 --- a/bmq-sdk/src/test/java/com/bloomberg/bmq/it/NettyTcpConnectionImplIT.java +++ b/bmq-sdk/src/test/java/com/bloomberg/bmq/it/NettyTcpConnectionImplIT.java @@ -979,6 +979,91 @@ void testChannelWaterMarkSlowServer() { logger.info("=============================================================="); } + @Test + void testWaitUntilWritableReturnsWhenChannelGoesDown() { + + logger.info("====================================================================="); + logger.info("BEGIN Testing 'waitUntilWritable' when the channel goes down."); + logger.info("====================================================================="); + + // A thread blocked in 'waitUntilWritable' must be released once the + // channel goes down, since a dead channel never becomes writable. + + // 1) Bring up the server and disable reads, so the client's write + // buffer cannot drain. + // 2) Invoke 'connect' and ensure that it succeeds. + // 3) Write messages until 'write' returns 'WRITE_BUFFER_FULL'. + // 4) Invoke 'waitUntilWritable' from another thread. + // 5) Stop the server, so the channel goes down while the thread is + // still blocked. + // 6) Ensure that the blocked thread is released. + + init(); + + final String message = "Payload for NettyTcpConnection integration test"; + + SessionOptions so = SessionOptions.builder().setBrokerUri(getServerUri()).build(); + + // 1) Bring up the server (only netty-based server supports disabling + // reads). + TestTcpServer server = new BmqBrokerSimulator(so.brokerUri().getPort(), Mode.SILENT_MODE); + server.start(); + + TcpConnection impl = NettyTcpConnection.createInstance(); + + impl.setChannelStatusHandler(eventHandler); + + // 2) Invoke 'connect' and ensure that it succeeds. + logger.info("Initiating connection..."); + int rc = + impl.connect( + new ConnectionOptions(so), eventHandler, eventHandler, MIN_NUM_READ_BYTES); + + assertEquals(0, rc); + TestTools.acquireSema(connectSema); + TestTools.acquireSema(channelUpSema); + + server.disableRead(); + + // 3) Write messages until 'write' returns 'WRITE_BUFFER_FULL'. + ByteBuffer packet = ByteBuffer.wrap(message.getBytes(StandardCharsets.US_ASCII)); + ByteBuffer[] data = new ByteBuffer[] {packet}; + WriteStatus writeRc; + while (true) { + writeRc = impl.write(data); + if (writeRc != WriteStatus.SUCCESS) { + break; + } + } + assertEquals(WriteStatus.WRITE_BUFFER_FULL, writeRc); + + // 4) Invoke 'waitUntilWritable' from another thread. It is a daemon + // so that it cannot keep the JVM alive if it is never released. + Thread waiter = new Thread(impl::waitUntilWritable, "waitUntilWritable_thread"); + waiter.setDaemon(true); + waiter.start(); + + TestTools.sleepForSeconds(1); + assertTrue(waiter.isAlive()); + + // 5) Stop the server, so the channel goes down while the thread is + // still blocked. + server.stop(); + TestTools.acquireSema(channelDownSema); + + // 6) Ensure that the blocked thread is released. + try { + waiter.join(Duration.ofSeconds(30).toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + assertFalse(waiter.isAlive()); + + logger.info("==================================================================="); + logger.info("END Testing 'waitUntilWritable' when the channel goes down."); + logger.info("==================================================================="); + } + @Test void testBmqServer() throws IOException { logger.info("======================================================================"); diff --git a/bmq-sdk/src/test/java/com/bloomberg/bmq/it/util/BmqBrokerSimulator.java b/bmq-sdk/src/test/java/com/bloomberg/bmq/it/util/BmqBrokerSimulator.java index 3ad8b784..5d8d8866 100644 --- a/bmq-sdk/src/test/java/com/bloomberg/bmq/it/util/BmqBrokerSimulator.java +++ b/bmq-sdk/src/test/java/com/bloomberg/bmq/it/util/BmqBrokerSimulator.java @@ -408,15 +408,21 @@ public void stop() { @Override public void enableRead() { - if (channelFuture != null) { - channelFuture.channel().config().setAutoRead(true); - } + setAutoRead(true); } @Override public void disableRead() { - if (channelFuture != null) { - channelFuture.channel().config().setAutoRead(false); + setAutoRead(false); + } + + private void setAutoRead(boolean value) { + // Applies to the accepted client connection, not to the listening + // socket: suppressing reads here stops draining the client's data and + // lets its write buffer grow. + ChannelHandlerContext ctx = channelContext; + if (ctx != null) { + ctx.channel().config().setAutoRead(value); } } From 01c4686a31e117998f6939cb414d6f777e51aaed Mon Sep 17 00:00:00 2001 From: Evgeny Malygin Date: Fri, 4 Sep 2026 12:11:38 -0400 Subject: [PATCH 2/2] Fix: avoid eternal lock on Semaphore Signed-off-by: Evgeny Malygin --- .../bmq/impl/infr/net/NettyTcpConnection.java | 48 +++++++++---------- 1 file changed, 24 insertions(+), 24 deletions(-) diff --git a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/net/NettyTcpConnection.java b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/net/NettyTcpConnection.java index 3ab37cde..226b0eee 100644 --- a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/net/NettyTcpConnection.java +++ b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/net/NettyTcpConnection.java @@ -43,7 +43,6 @@ import java.net.InetSocketAddress; import java.net.URI; import java.nio.ByteBuffer; -import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -199,7 +198,6 @@ private enum ChannelState { private ConnectCallback connectCallback; private DisconnectCallback disconnectCallback; private ChannelStatusHandler channelStatusHandler; - private Semaphore channelWaterMarkSema; private final ClientChannelAdapter clientChannelAdapter; private final long lingerTimeout; @@ -335,7 +333,6 @@ public int connect( readBuffer = new ByteBufferOutputStream(); readBytesStatus = new ReadCompletionStatus(); readBytesStatus.setNumNeeded(initialMinNumBytes); - channelWaterMarkSema = new Semaphore(0); doConnect(); } @@ -373,7 +370,7 @@ public int disconnect(DisconnectCallback disconnectCb) { // Unblock any thread waiting for the channel to become writable // (client could invoke 'disconnect' from one thread, while another // thread is blocked on 'waitUntilWritable'). - channelWaterMarkSema.release(); + lock.notifyAll(); logger.debug("disconnect in state {}", state); @@ -555,7 +552,11 @@ public boolean isWritable() { } /** - * Wait until channel becomes writable. + * Wait until channel becomes writable, or until it goes down. + * + *

A channel which is no longer connected never becomes writable, so the wait ends when the + * channel does. Waiters are notified under {@code lock}, which also guards the waited-for + * state, so a notification cannot be missed by a thread which has not blocked yet. * *

Thread model: executed in any thread except I/O thread. * @@ -564,26 +565,20 @@ public boolean isWritable() { @SuppressWarnings("squid:S2142") public void waitUntilWritable() { - Semaphore sema = null; - synchronized (lock) { if (Thread.currentThread().getId() == ioThreadId) { throw new IllegalStateException( "Cannot invoke 'waitUntilWritable' from the IO thread."); } - if (state == ChannelState.CONNECTED && !channelContext.channel().isWritable()) { - sema = channelWaterMarkSema; - } // else: channel is not connected, or is writable. - } - - // Wait indefinitely on the semaphore outside the lock. - if (sema != null) { - try { - sema.acquire(); - } catch (InterruptedException e) { - logger.info("InterruptedException: ", e); - Thread.currentThread().interrupt(); + while (state == ChannelState.CONNECTED && !channelContext.channel().isWritable()) { + try { + lock.wait(); + } catch (InterruptedException e) { + logger.info("InterruptedException: ", e); + Thread.currentThread().interrupt(); + return; + } } } } @@ -734,9 +729,9 @@ public void channelInactive(ChannelHandlerContext ctx) { */ @Override public void channelWritabilityChanged(ChannelHandlerContext ctx) { - // If channel has become writable, post on the semaphore on which - // 'write' might be waiting, and also send CHANNEL_WRITABLE status. - // Else, simply log. + // If channel has become writable, wake the threads which might be + // waiting for it in 'waitUntilWritable', and also send + // CHANNEL_WRITABLE status. Else, simply log. logger.info( "channelWritabilityChanged. isWritable: {}, BytesBeforeUnwritable: {}, BytesBeforeWritable: {}", @@ -748,8 +743,8 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) { ChannelStatusHandler channelHandler = null; synchronized (lock) { channelHandler = channelStatusHandler; + lock.notifyAll(); } - channelWaterMarkSema.release(); if (channelHandler != null) { channelHandler.handleChannelStatus(ChannelStatus.CHANNEL_WRITABLE); } @@ -843,6 +838,11 @@ private void channelCloseFutureComplete() { logger.debug("channelCloseFutureComplete {}", state); + // The channel is gone, so it can never become writable again. + // Release the threads waiting for that to happen; their writes + // complete with a not-connected result. + lock.notifyAll(); + if (state == ChannelState.DISCONNECTING) { // Disconnection complete. @@ -884,7 +884,7 @@ private void connectFutureComplete(ChannelFuture future) { if (future.isSuccess()) { future.channel().closeFuture().addListener(this); future.channel().close(); - // We will post on the disconnecting-semaphore in the + // The disconnect callback is invoked from the // channel-close future. } // else: future is cancelled or failure, in which case, there