Skip to content

Fix: Improper read timeout handling. #621

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

Closed
wants to merge 2 commits into from
Closed

Fix: Improper read timeout handling. #621

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
dce10be
Fix: Improper read timeout handling.
konrad-kaminski Dec 28, 2020
708bb59
Fix: Improper read timeout handling.
konrad-kaminski Jan 14, 2021
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
107 changes: 52 additions & 55 deletions driver-core/src/main/com/mongodb/connection/netty/NettyStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
Expand All @@ -44,7 +43,6 @@
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.ReadTimeoutException;
import io.netty.util.concurrent.EventExecutor;
import org.bson.ByteBuf;

import javax.net.ssl.SSLContext;
Expand All @@ -58,6 +56,8 @@
import java.util.List;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static com.mongodb.internal.connection.SslHelper.enableHostNameVerification;
import static com.mongodb.internal.connection.SslHelper.enableSni;
Expand All @@ -67,7 +67,7 @@
* A Stream implementation based on Netty 4.0.
*/
final class NettyStream implements Stream {
private static final String READ_HANDLER_NAME = "ReadTimeoutHandler";
private static final String INBOUND_BUFFER_HANDLER_NAME = "InboundBufferHandler";
private final ServerAddress address;
private final SocketSettings settings;
private final SslSettings sslSettings;
Expand All @@ -81,6 +81,7 @@ final class NettyStream implements Stream {
private final LinkedList<io.netty.buffer.ByteBuf> pendingInboundBuffers = new LinkedList<io.netty.buffer.ByteBuf>();
private volatile PendingReader pendingReader;
private volatile Throwable pendingException;
private final int readTimeoutMs;

NettyStream(final ServerAddress address, final SocketSettings settings, final SslSettings sslSettings, final EventLoopGroup workerGroup,
final Class<? extends SocketChannel> socketChannelClass, final ByteBufAllocator allocator) {
Expand All @@ -90,6 +91,7 @@ final class NettyStream implements Stream {
this.workerGroup = workerGroup;
this.socketChannelClass = socketChannelClass;
this.allocator = allocator;
this.readTimeoutMs = settings.getReadTimeout(MILLISECONDS);
}

@Override
Expand Down Expand Up @@ -154,11 +156,7 @@ public void initChannel(final SocketChannel ch) {
engine.setSSLParameters(sslParameters);
ch.pipeline().addFirst("ssl", new SslHandler(engine, false));
}
int readTimeout = settings.getReadTimeout(MILLISECONDS);
if (readTimeout > 0) {
ch.pipeline().addLast(READ_HANDLER_NAME, new ReadTimeoutHandler(readTimeout));
}
ch.pipeline().addLast(new InboundBufferHandler());
ch.pipeline().addLast(INBOUND_BUFFER_HANDLER_NAME, new InboundBufferHandler());
}
});
final ChannelFuture channelFuture = bootstrap.connect(nextAddress);
Expand Down Expand Up @@ -186,7 +184,7 @@ public boolean supportsAdditionalTimeout() {
@Override
public ByteBuf read(final int numBytes, final int additionalTimeout) throws IOException {
FutureAsyncCompletionHandler<ByteBuf> future = new FutureAsyncCompletionHandler<ByteBuf>();
readAsync(numBytes, future, additionalTimeout);
readAsync(numBytes, future, additionalTimeout, null);
return future.get();
}

Expand All @@ -211,18 +209,19 @@ public void operationComplete(final ChannelFuture future) throws Exception {

@Override
public void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf> handler) {
readAsync(numBytes, handler, 0);
readAsync(numBytes, handler, 0, null);
}

private void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf> handler, final int additionalTimeout) {
scheduleReadTimeout(additionalTimeout);
private void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf> handler, final int additionalTimeout,
final PendingReader pendingReaderToReuse) {
ByteBuf buffer = null;
Throwable exceptionResult = null;
Throwable exceptionResult;
synchronized (this) {
exceptionResult = pendingException;
if (exceptionResult == null) {
if (!hasBytesAvailable(numBytes)) {
pendingReader = new PendingReader(numBytes, handler);
pendingReader = (pendingReaderToReuse != null) ? pendingReaderToReuse
: new PendingReader(numBytes, handler, scheduleReadTimeout(additionalTimeout));
} else {
CompositeByteBuf composite = allocator.compositeBuffer(pendingInboundBuffers.size());
int bytesNeeded = numBytes;
Expand All @@ -246,12 +245,17 @@ private void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf>
}
}
}

if ((exceptionResult != null || buffer != null)
&& (pendingReaderToReuse != null)
&& (pendingReaderToReuse.timeoutFuture != null)) {
pendingReaderToReuse.timeoutFuture.cancel(false);
}

if (exceptionResult != null) {
disableReadTimeout();
handler.failed(exceptionResult);
}
if (buffer != null) {
disableReadTimeout();
handler.completed(buffer);
}
}
Expand Down Expand Up @@ -282,7 +286,7 @@ private void handleReadResponse(final io.netty.buffer.ByteBuf buffer, final Thro
}

if (localPendingReader != null) {
readAsync(localPendingReader.numBytes, localPendingReader.handler);
readAsync(localPendingReader.numBytes, localPendingReader.handler, 0, localPendingReader);
}
}

Expand Down Expand Up @@ -358,10 +362,12 @@ public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable t)
private static final class PendingReader {
private final int numBytes;
private final AsyncCompletionHandler<ByteBuf> handler;
private final Future<?> timeoutFuture;

private PendingReader(final int numBytes, final AsyncCompletionHandler<ByteBuf> handler) {
private PendingReader(final int numBytes, final AsyncCompletionHandler<ByteBuf> handler, final Future<?> timeoutFuture) {
this.numBytes = numBytes;
this.handler = handler;
this.timeoutFuture = timeoutFuture;
}
}

Expand Down Expand Up @@ -445,47 +451,38 @@ public void operationComplete(final ChannelFuture future) {
}
}

private void scheduleReadTimeout(final int additionalTimeout) {
adjustTimeout(false, additionalTimeout);
}
private Future<?> scheduleReadTimeout(final int additionalTimeout) {
if (isClosed || readTimeoutMs <= 0) {
return null;
}

private void disableReadTimeout() {
adjustTimeout(true, 0);
final ChannelHandlerContext ctx = channel.pipeline().context(INBOUND_BUFFER_HANDLER_NAME);
if (ctx != null) {
final ReadTimeoutTask task = new ReadTimeoutTask(ctx);
return ctx.executor().schedule(task, readTimeoutMs + additionalTimeout, TimeUnit.MILLISECONDS);
} else {
return null;
}
}

private void adjustTimeout(final boolean disable, final int additionalTimeout) {
if (isClosed) {
return;
}
ChannelHandler timeoutHandler = channel.pipeline().get(READ_HANDLER_NAME);
if (timeoutHandler != null) {
final ReadTimeoutHandler readTimeoutHandler = (ReadTimeoutHandler) timeoutHandler;
final ChannelHandlerContext handlerContext = channel.pipeline().context(timeoutHandler);
EventExecutor executor = handlerContext.executor();

if (disable) {
if (executor.inEventLoop()) {
readTimeoutHandler.removeTimeout(handlerContext);
} else {
executor.submit(new Runnable() {
@Override
public void run() {
readTimeoutHandler.removeTimeout(handlerContext);
}
});
}
} else {
if (executor.inEventLoop()) {
readTimeoutHandler.scheduleTimeout(handlerContext, additionalTimeout);
} else {
executor.submit(new Runnable() {
@Override
public void run() {
readTimeoutHandler.scheduleTimeout(handlerContext, additionalTimeout);
}
});
}
private static final class ReadTimeoutTask implements Runnable {

private final ChannelHandlerContext ctx;

ReadTimeoutTask(final ChannelHandlerContext ctx) {
this.ctx = ctx;
}

@Override
public void run() {
if (ctx.channel().isOpen()) {
try {
ctx.fireExceptionCaught(ReadTimeoutException.INSTANCE);
ctx.close();
} catch (Throwable t) {
ctx.fireExceptionCaught(t);
}
}
}
}
}
78 changes: 0 additions & 78 deletions driver-core/src/main/com/mongodb/connection/netty/ReadTimeoutHandler.java
View file
Open in desktop

This file was deleted.