Skip to content

JAVA-6071 use trampoline pattern to avoid stackoverflow in reactive driver #1973

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

Merged
strogiyotec merged 10 commits into from
Oct 14, 2026
Merged

JAVA-6071 use trampoline pattern to avoid stackoverflow in reactive driver #1973

Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
bfbaa96
JAVA-6071
strogiyotec May 18, 2026
396082b
Merge branch 'main' into JAVA-6071
strogiyotec Jun 16, 2026
92a2afa
Version: bump 5.11.0-SNAPSHOT
strogiyotec Aug 13, 2026
170f6f0
Merge branch 'main' of https://github.com/mongodb/mongo-java-driver
strogiyotec Aug 18, 2026
92883f6
Merge branch 'main' of https://github.com/mongodb/mongo-java-driver
strogiyotec Sep 4, 2026
74f46b4
Merge branch 'main' of https://github.com/mongodb/mongo-java-driver
strogiyotec Sep 14, 2026
c723c39
Merge branch 'main' of https://github.com/mongodb/mongo-java-driver
strogiyotec Sep 24, 2026
b1b912a
Merge branch 'main' of https://github.com/mongodb/mongo-java-driver
strogiyotec Sep 26, 2026
e210fa9
Merge branch 'main' into JAVA-6071
strogiyotec Oct 1, 2026
514637d
Merge branch 'main' into JAVA-6071
strogiyotec Oct 14, 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
116 changes: 43 additions & 73 deletions driver-core/src/main/com/mongodb/internal/connection/AsynchronousChannelStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import com.mongodb.ServerAddress;
import com.mongodb.connection.AsyncCompletionHandler;
import com.mongodb.connection.SocketSettings;
import com.mongodb.internal.async.SingleResultCallback;
import com.mongodb.lang.Nullable;
import org.bson.ByteBuf;

Expand Down Expand Up @@ -92,31 +93,54 @@ public void writeAsync(final List<ByteBuf> buffers, final OperationContext opera
final AsyncCompletionHandler<Void> handler) {
AsyncWritableByteChannelAdapter byteChannel = new AsyncWritableByteChannelAdapter();

@vbabanin vbabanin Oct 8, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

[nit] The new code uses thenRunWhileLoop(iter::hasNext, ...), and a while loop checks its condition before running the body. With an empty list, the condition is false on the first check, so the body never runs and the method reports the write as completed even though no message was sent.

This does not look reachable through the driver’s normal code paths, so I do not think it is a user-visible. But it is an edge-case behavior change from the rewrite: loud failure before from iter.next();, silent no-op success now.

Could we add a small guard?

Suggested change
AsyncWritableByteChannelAdapter byteChannel = new AsyncWritableByteChannelAdapter();
assertTrue(!buffers.isEmpty())
AsyncWritableByteChannelAdapter byteChannel = new AsyncWritableByteChannelAdapter();

Iterator<ByteBuf> iter = buffers.iterator();
pipeOneBuffer(byteChannel, iter.next(), operationContext, new AsyncCompletionHandler<Void>() {
beginAsync().thenRunWhileLoop(
iter::hasNext,
c -> pipeOneBuffer(byteChannel, iter.next(), operationContext, c.asHandler())
).finish(handler.asCallback());
}

@Override
public void readAsync(final int numBytes, final OperationContext operationContext, final AsyncCompletionHandler<ByteBuf> handler) {
ByteBuf buffer = bufferProvider.getBuffer(numBytes);

beginAsync().thenRunWhileLoop(
buffer::hasRemaining,
c -> readOnce(buffer, operationContext, c)
).<ByteBuf>thenSupply(c -> {
buffer.flip();
c.complete(buffer);
}).onErrorIf(t -> true, (t, c) -> {
buffer.release();
if (t instanceof InterruptedByTimeoutException) {
c.completeExceptionally(new MongoSocketReadTimeoutException(
"Timeout while receiving message", serverAddress, t));
} else {
c.completeExceptionally(t);
}
}).finish(handler.asCallback());
}

private void readOnce(final ByteBuf buffer, final OperationContext operationContext, final SingleResultCallback<Void> callback) {
long readTimeoutMS = operationContext.getTimeoutContext().getReadTimeoutMS();
getChannel().read(buffer.asNIO(), readTimeoutMS, MILLISECONDS, null, new BaseCompletionHandler<Void, Integer, Void>(callback.asHandler()) {
@Override
public void completed(@Nullable final Void t) {
if (iter.hasNext()) {
pipeOneBuffer(byteChannel, iter.next(), operationContext, this);
public void completed(final Integer result, final Void attachment) {
AsyncCompletionHandler<Void> localHandler = getHandlerAndClear();
if (result == -1) {
localHandler.failed(new MongoSocketReadException("Prematurely reached end of stream", serverAddress));
} else {
handler.completed(null);
localHandler.completed(null);
}
}

@Override
public void failed(final Throwable t) {
handler.failed(t);
public void failed(final Throwable t, final Void attachment) {
AsyncCompletionHandler<Void> localHandler = getHandlerAndClear();
localHandler.failed(t);
}
});
}

@Override
public void readAsync(final int numBytes, final OperationContext operationContext, final AsyncCompletionHandler<ByteBuf> handler) {
ByteBuf buffer = bufferProvider.getBuffer(numBytes);

long timeout = operationContext.getTimeoutContext().getReadTimeoutMS();
getChannel().read(buffer.asNIO(), timeout, MILLISECONDS, null, new BasicCompletionHandler(buffer, operationContext, handler));
}

@Override
public void open(final OperationContext operationContext) throws IOException {
FutureAsyncCompletionHandler<Void> handler = new FutureAsyncCompletionHandler<>();
Expand Down Expand Up @@ -171,21 +195,10 @@ public ByteBuf getBuffer(final int size) {

private void pipeOneBuffer(final AsyncWritableByteChannelAdapter byteChannel, final ByteBuf byteBuffer,
final OperationContext operationContext, final AsyncCompletionHandler<Void> outerHandler) {
byteChannel.write(byteBuffer.asNIO(), operationContext, new AsyncCompletionHandler<Void>() {
@Override
public void completed(@Nullable final Void t) {
if (byteBuffer.hasRemaining()) {
byteChannel.write(byteBuffer.asNIO(), operationContext, this);
} else {
outerHandler.completed(null);
}
}

@Override
public void failed(final Throwable t) {
outerHandler.failed(t);
}
});
beginAsync().thenRunWhileLoop(
byteBuffer::hasRemaining,
c -> byteChannel.write(byteBuffer.asNIO(), operationContext, c.asHandler())
).finish(outerHandler.asCallback());
}

private class AsyncWritableByteChannelAdapter {
Expand Down Expand Up @@ -221,49 +234,6 @@ public void failed(final Throwable t, final Object attachment) {
}
}

private final class BasicCompletionHandler extends BaseCompletionHandler<ByteBuf, Integer, Void> {
private final AtomicReference<ByteBuf> byteBufReference;
private final OperationContext operationContext;

private BasicCompletionHandler(final ByteBuf dst, final OperationContext operationContext,
final AsyncCompletionHandler<ByteBuf> handler) {
super(handler);
this.byteBufReference = new AtomicReference<>(dst);
this.operationContext = operationContext;
}

@Override
public void completed(final Integer result, final Void attachment) {
AsyncCompletionHandler<ByteBuf> localHandler = getHandlerAndClear();
beginAsync().<ByteBuf>thenSupply((c) -> {
ByteBuf localByteBuf = byteBufReference.getAndSet(null);
if (result == -1) {
localByteBuf.release();
throw new MongoSocketReadException("Prematurely reached end of stream", serverAddress);
} else if (!localByteBuf.hasRemaining()) {
localByteBuf.flip();
c.complete(localByteBuf);
} else {
long readTimeoutMS = operationContext.getTimeoutContext().getReadTimeoutMS();
getChannel().read(localByteBuf.asNIO(), readTimeoutMS, MILLISECONDS, null,
new BasicCompletionHandler(localByteBuf, operationContext, c.asHandler()));
}
}).finish(localHandler.asCallback());
}

@Override
public void failed(final Throwable t, final Void attachment) {
AsyncCompletionHandler<ByteBuf> localHandler = getHandlerAndClear();
ByteBuf localByteBuf = byteBufReference.getAndSet(null);
localByteBuf.release();
if (t instanceof InterruptedByTimeoutException) {
localHandler.failed(new MongoSocketReadTimeoutException("Timeout while receiving message", serverAddress, t));
} else {
localHandler.failed(t);
}
}
}

// Private base class for all CompletionHandler implementors that ensures the upstream handler is
// set to null before it is used. This is to work around an observed issue with implementations of
// AsynchronousSocketChannel that fail to clear references to handlers stored in instance fields of
Expand Down