Skip to content

Fix: replace channel semaphore with a lock #135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
678098 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix: replace channel semaphore with a lock #135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5485b49
IT[NettyTcpConnectionImplIT]: add is writeable test
678098 Sep 4, 2026
01c4686
Fix: avoid eternal lock on Semaphore
678098 Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 24 additions & 24 deletions bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/net/NettyTcpConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -335,7 +333,6 @@ public int connect(
readBuffer = new ByteBufferOutputStream();
readBytesStatus = new ReadCompletionStatus();
readBytesStatus.setNumNeeded(initialMinNumBytes);
channelWaterMarkSema = new Semaphore(0);
doConnect();
}

Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -555,7 +552,11 @@ public boolean isWritable() {
}

/**
* Wait until channel becomes writable.
* Wait until channel becomes writable, or until it goes down.
*
* <p>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.
*
* <p>Thread model: executed in any thread except I/O thread.
*
Expand All @@ -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;
}
}
}
}
Expand Down Expand Up @@ -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: {}",
Expand All @@ -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);
}
Expand Down Expand Up @@ -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();

@678098 678098 Sep 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The original code hasn't got any semaphore notification during channel close.
This meant that a thread that is blocked in waitUntilWritable will never be unblocked if a channel was closed.

We don't even need a separate semaphore to send this notification: existing lock is enough.


if (state == ChannelState.DISCONNECTING) {
// Disconnection complete.

Expand Down Expand Up @@ -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
Expand Down
85 changes: 85 additions & 0 deletions bmq-sdk/src/test/java/com/bloomberg/bmq/it/NettyTcpConnectionImplIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());

@678098 678098 Sep 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fails in main:

[ERROR] com.bloomberg.bmq.it.NettyTcpConnectionImplIT.testWaitUntilWritableReturnsWhenChannelGoesDown -- Time elapsed: 31.59 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <false> but was: <true>
	at org.junit.jupiter.api.Assertions.assertFalse(Assertions.java:257)
	at com.bloomberg.bmq.it.NettyTcpConnectionImplIT.testWaitUntilWritableReturnsWhenChannelGoesDown(NettyTcpConnectionImplIT.java:1060)


logger.info("===================================================================");
logger.info("END Testing 'waitUntilWritable' when the channel goes down.");
logger.info("===================================================================");
}

@Test
void testBmqServer() throws IOException {
logger.info("======================================================================");
Expand Down
16 changes: 11 additions & 5 deletions bmq-sdk/src/test/java/com/bloomberg/bmq/it/util/BmqBrokerSimulator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}

Expand Down
Loading