From c0bcc26f10c65cbcda9ab033f84ecb9993986907 Mon Sep 17 00:00:00 2001 From: Valentin Kovalenko Date: Fri, 15 Jan 2021 22:00:51 -0700 Subject: [PATCH 1/4] Fix the read timeout implementation in NettyStream Both the new approach and the original one achieve the guarantee that there are no concurrent read timeouts scheduled for a channel. This is an essential property needed for a timeout implementation, let us call it "at-most-one". The original approach of achieving the at-most-one property: - Schedule timeouts only by an event loop thread. - If another thread needs to schedule a timeout, it submits a new scheduleTimeout task to the channel's event loop (asynchronous timeout scheduling). - The scheduleTimeout task schedules a new timeout if none is scheduled. The original approach allowed executions in which a scheduleTimeout task runs after completion of the read operation that submitted the task, which resulted in unexpected timeout exceptions. The new approach achieves the at-most-one property by using a lock. As a result, timeouts can be scheduled by any thread and there is no asynchronous timeout scheduling. This means we cannot miss removing a timeout because it was submitted for scheduling, but has not been scheduled yet. Other notable changes: - [related bug fix] NettyStream now must always respect the requested additional timeout. The original implementation had a chance to ignore the requested additional timeout and schedule a timeout with the default delay. This is again due to the asynchronous timeout scheduling in the original approach. - [performance optimization] Public read methods do not schedule timeouts anymore if the requested number of bytes is available right away. - [performance optimization] Netty channel handlers do not try to schedule timeouts anymore, timeouts may be scheduled only by the public read methods. Trying to schedule timeouts from the method handleReadResponse was unnecessary even in the original approach. - [performance optimization] NettyStream does not produce excessive garbage by re-creating PendingReader objects each time a new piece of data arrives for a pending reader. - [code improvement] The fields NettyStream.pendingReader, pendingException are always written/read inside synchronized blocks that use the same NettyStream object, so they can be plain. Marking them volatile is unnecessary and potentially misleading. - [code improvement] ReadTimeoutHandler was removed because it wasn't acting as a handler and was not needed. JAVA-3920 --- .../mongodb/connection/netty/NettyStream.java | 200 ++++++++++++------ .../connection/netty/ReadTimeoutHandler.java | 78 ------- 2 files changed, 141 insertions(+), 137 deletions(-) delete mode 100644 driver-core/src/main/com/mongodb/connection/netty/ReadTimeoutHandler.java diff --git a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java index 948da9162ec..212c2e4ac37 100644 --- a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java +++ b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java @@ -24,10 +24,12 @@ import com.mongodb.MongoSocketOpenException; import com.mongodb.MongoSocketReadTimeoutException; import com.mongodb.ServerAddress; +import com.mongodb.annotations.ThreadSafe; import com.mongodb.connection.AsyncCompletionHandler; import com.mongodb.connection.SocketSettings; import com.mongodb.connection.SslSettings; import com.mongodb.connection.Stream; +import com.mongodb.lang.Nullable; import io.netty.bootstrap.Bootstrap; import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.CompositeByteBuf; @@ -35,16 +37,16 @@ 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.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; +import io.netty.channel.ChannelPipeline; import io.netty.channel.EventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; 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; @@ -58,16 +60,49 @@ import java.util.List; import java.util.Queue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledFuture; +import static com.mongodb.assertions.Assertions.isTrueArgument; import static com.mongodb.internal.connection.SslHelper.enableHostNameVerification; import static com.mongodb.internal.connection.SslHelper.enableSni; import static java.util.concurrent.TimeUnit.MILLISECONDS; /** * A Stream implementation based on Netty 4.0. + * Just like it is for the {@link java.nio.channels.AsynchronousSocketChannel}, + * concurrent pending1 readers + * (whether {@linkplain #read(int, int) synchronous} or {@linkplain #readAsync(int, AsyncCompletionHandler) asynchronous}) + * are not supported by {@link NettyStream}. + * However, this class does not have a fail-fast mechanism checking for such situations. + *
+ * 1We cannot simply say that read methods are not allowed be run concurrently because strictly speaking they are allowed, + * as explained below. + * 
{@code
+ * NettyStream stream = ...;
+ * stream.readAsync(1, new AsyncCompletionHandler() {//inv1
+ *  @Override
+ *  public void completed(ByteBuf o) {
+ *      stream.readAsync(//inv2
+ *              1, ...);//ret2
+ *  }
+ *
+ *  @Override
+ *  public void failed(Throwable t) {
+ *  }
+ * });//ret1
+ * }
+ * Arrows on the diagram below represent happens-before relations. + * 
{@code
+ * int1 -> inv2 -> ret2
+ *      \--------> ret1
+ * }
+ * As shown on the diagram, the method {@link #readAsync(int, AsyncCompletionHandler)} runs concurrently with + * itself in the example above. However, there are no concurrent pending readers because the second operation + * is invoked after the first operation has completed reading despite the method has not returned yet. */ final class NettyStream implements Stream { - private static final String READ_HANDLER_NAME = "ReadTimeoutHandler"; + private static final int NO_SCHEDULE_TIMEOUT = -1; private final ServerAddress address; private final SocketSettings settings; private final SslSettings sslSettings; @@ -75,12 +110,22 @@ final class NettyStream implements Stream { private final Class socketChannelClass; private final ByteBufAllocator allocator; - private volatile boolean isClosed; + private boolean isClosed; private volatile Channel channel; private final LinkedList pendingInboundBuffers = new LinkedList(); - private volatile PendingReader pendingReader; - private volatile Throwable pendingException; + /* The fields pendingReader, pendingException are always written/read inside synchronized blocks + * that use the same NettyStream object, so they can be plain.*/ + private PendingReader pendingReader; + private Throwable pendingException; + /* The fields readTimeoutTask, readTimeoutMillis are each written only in the ChannelInitializer.initChannel method + * (in addition to the write of the default value), and read only when NettyStream users read data, + * or Netty event loop handles incoming data. Since actions done by the ChannelInitializer.initChannel method + * are ordered (in the happens-before order) before user read actions and before event loop actions that handle incoming data, + * these fields can be plain.*/ + @Nullable + private ReadTimeoutTask readTimeoutTask; + private long readTimeoutMillis; NettyStream(final ServerAddress address, final SocketSettings settings, final SslSettings sslSettings, final EventLoopGroup workerGroup, final Class socketChannelClass, final ByteBufAllocator allocator) { @@ -143,6 +188,7 @@ private void initializeChannel(final AsyncCompletionHandler handler, final bootstrap.handler(new ChannelInitializer() { @Override public void initChannel(final SocketChannel ch) { + final ChannelPipeline pipeline = ch.pipeline(); if (sslSettings.isEnabled()) { SSLEngine engine = getSslContext().createSSLEngine(address.getHost(), address.getPort()); engine.setUseClientMode(true); @@ -152,13 +198,24 @@ public void initChannel(final SocketChannel ch) { enableHostNameVerification(sslParameters); } engine.setSSLParameters(sslParameters); - ch.pipeline().addFirst("ssl", new SslHandler(engine, false)); + pipeline.addFirst("ssl", new SslHandler(engine, false)); } + int readTimeout = settings.getReadTimeout(MILLISECONDS); + final long readTimeoutMillis; if (readTimeout > 0) { - ch.pipeline().addLast(READ_HANDLER_NAME, new ReadTimeoutHandler(readTimeout)); + readTimeoutMillis = readTimeout; + /* We need at least one handler before (in the inbound evaluation order) the InboundBufferHandler, + * so that we can fire exception events (they are inbound events) using its context and the InboundBufferHandler + * receives them. SslHandler is not always present, so adding a NOOP handler.*/ + pipeline.addLast(new ChannelInboundHandlerAdapter()); + readTimeoutTask = new ReadTimeoutTask(pipeline.lastContext()); + } else { + readTimeoutMillis = NO_SCHEDULE_TIMEOUT; } - ch.pipeline().addLast(new InboundBufferHandler()); + NettyStream.this.readTimeoutMillis = readTimeoutMillis; + + pipeline.addLast(new InboundBufferHandler()); } }); final ChannelFuture channelFuture = bootstrap.connect(nextAddress); @@ -184,9 +241,10 @@ public boolean supportsAdditionalTimeout() { } @Override - public ByteBuf read(final int numBytes, final int additionalTimeout) throws IOException { + public ByteBuf read(final int numBytes, final int additionalTimeoutMillis) throws IOException { + isTrueArgument("additionalTimeoutMillis must not be negative", additionalTimeoutMillis >= 0); FutureAsyncCompletionHandler future = new FutureAsyncCompletionHandler(); - readAsync(numBytes, future, additionalTimeout); + readAsync(numBytes, future, combinedTimeout(readTimeoutMillis, additionalTimeoutMillis)); return future.get(); } @@ -211,18 +269,32 @@ public void operationComplete(final ChannelFuture future) throws Exception { @Override public void readAsync(final int numBytes, final AsyncCompletionHandler handler) { - readAsync(numBytes, handler, 0); + readAsync(numBytes, handler, readTimeoutMillis); } - private void readAsync(final int numBytes, final AsyncCompletionHandler handler, final int additionalTimeout) { - scheduleReadTimeout(additionalTimeout); + /** + * @param numBytes Must be equal to {@link #pendingReader}{@code .numBytes} when called by a Netty channel handler. + * @param handler Must be equal to {@link #pendingReader}{@code .handler} when called by a Netty channel handler. + * @param readTimeoutMillis Must be equal to {@link #NO_SCHEDULE_TIMEOUT} when called by a Netty channel handler. + * Timeouts may be scheduled only by the public read methods. Taking into account that concurrent pending + * readers are not allowed, there must not be a situation when threads attempt to schedule a timeout + * before the previous one is either cancelled or completed. + */ + private void readAsync(final int numBytes, final AsyncCompletionHandler handler, final long readTimeoutMillis) { ByteBuf buffer = null; Throwable exceptionResult = null; synchronized (this) { exceptionResult = pendingException; + PendingReader pendingReader = this.pendingReader; if (exceptionResult == null) { if (!hasBytesAvailable(numBytes)) { - pendingReader = new PendingReader(numBytes, handler); + if (pendingReader == null) {//called by a public read method + pendingReader = new PendingReader(numBytes, handler, scheduleReadTimeout(readTimeoutTask, readTimeoutMillis)); + this.pendingReader = pendingReader; + } else {//called by a Netty channel handler + //assert pendingReader.numBytes == numBytes : "Concurrent pending readers are not allowed"; + //assert pendingReader.handler == handler : "Concurrent pending readers are not allowed"; + } } else { CompositeByteBuf composite = allocator.compositeBuffer(pendingInboundBuffers.size()); int bytesNeeded = numBytes; @@ -245,13 +317,16 @@ private void readAsync(final int numBytes, final AsyncCompletionHandler buffer = new NettyByteBuf(composite).flip(); } } + if (!(exceptionResult == null && buffer == null)//the read operation has completed + && pendingReader != null) {//we need to clear the pending reader + cancel(pendingReader.timeout); + this.pendingReader = null; + } } if (exceptionResult != null) { - disableReadTimeout(); handler.failed(exceptionResult); } if (buffer != null) { - disableReadTimeout(); handler.completed(buffer); } } @@ -275,14 +350,12 @@ private void handleReadResponse(final io.netty.buffer.ByteBuf buffer, final Thro } else { pendingException = t; } - if (pendingReader != null) { - localPendingReader = pendingReader; - pendingReader = null; - } + localPendingReader = pendingReader; } if (localPendingReader != null) { - readAsync(localPendingReader.numBytes, localPendingReader.handler); + //timeouts may be scheduled only by the public read methods + readAsync(localPendingReader.numBytes, localPendingReader.handler, NO_SCHEDULE_TIMEOUT); } } @@ -358,10 +431,14 @@ public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable t) private static final class PendingReader { private final int numBytes; private final AsyncCompletionHandler handler; + @Nullable + private final ScheduledFuture timeout; - private PendingReader(final int numBytes, final AsyncCompletionHandler handler) { + private PendingReader( + final int numBytes, final AsyncCompletionHandler handler, @Nullable final ScheduledFuture timeout) { this.numBytes = numBytes; this.handler = handler; + this.timeout = timeout; } } @@ -445,47 +522,52 @@ public void operationComplete(final ChannelFuture future) { } } - private void scheduleReadTimeout(final int additionalTimeout) { - adjustTimeout(false, additionalTimeout); + private static void cancel(@Nullable final Future f) { + if (f != null) { + f.cancel(false); + } } - private void disableReadTimeout() { - adjustTimeout(true, 0); + private static long combinedTimeout(final long timeout, final int additionalTimeout) { + if (timeout == NO_SCHEDULE_TIMEOUT || additionalTimeout == NO_SCHEDULE_TIMEOUT) { + return NO_SCHEDULE_TIMEOUT; + } else { + return Math.addExact(timeout, additionalTimeout); + } } - 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 ScheduledFuture scheduleReadTimeout(@Nullable final ReadTimeoutTask readTimeoutTask, final long timeoutMillis) { + if (timeoutMillis == NO_SCHEDULE_TIMEOUT) { + return null; + } else { + //assert readTimeoutTask != null : "readTimeoutTask must be initialized if read timeouts are enabled"; + return readTimeoutTask.schedule(timeoutMillis); + } + } + + @ThreadSafe + private static final class ReadTimeoutTask implements Runnable { + private final ChannelHandlerContext ctx; + + private ReadTimeoutTask(final ChannelHandlerContext timeoutChannelHandlerContext) { + ctx = timeoutChannelHandlerContext; + } + + @Override + public void run() { + try { + if (ctx.channel().isOpen()) { + ctx.fireExceptionCaught(ReadTimeoutException.INSTANCE); + ctx.close(); } + } catch (final Throwable t) { + ctx.fireExceptionCaught(t); } + } + + private ScheduledFuture schedule(final long timeoutMillis) { + //assert timeoutMillis > 0 : timeoutMillis; + return ctx.executor().schedule(this, timeoutMillis, MILLISECONDS); + } } } diff --git a/driver-core/src/main/com/mongodb/connection/netty/ReadTimeoutHandler.java b/driver-core/src/main/com/mongodb/connection/netty/ReadTimeoutHandler.java deleted file mode 100644 index 824c8f7d6a3..00000000000 --- a/driver-core/src/main/com/mongodb/connection/netty/ReadTimeoutHandler.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Copyright 2008-present MongoDB, Inc. - * Copyright 2012 The Netty Project - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package com.mongodb.connection.netty; - -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.handler.timeout.ReadTimeoutException; - -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; - -import static com.mongodb.assertions.Assertions.isTrue; -import static com.mongodb.assertions.Assertions.isTrueArgument; - -/** - * Passes a {@link ReadTimeoutException} if the time between a {@link #scheduleTimeout} and {@link #removeTimeout} is longer than the set - * timeout. - */ -final class ReadTimeoutHandler extends ChannelInboundHandlerAdapter { - private final long readTimeout; - private volatile ScheduledFuture timeout; - - ReadTimeoutHandler(final long readTimeout) { - isTrueArgument("readTimeout must be greater than zero.", readTimeout > 0); - this.readTimeout = readTimeout; - } - - void scheduleTimeout(final ChannelHandlerContext ctx, final int additionalTimeout) { - isTrue("Handler called from the eventLoop", ctx.channel().eventLoop().inEventLoop()); - if (timeout == null) { - timeout = ctx.executor().schedule(new ReadTimeoutTask(ctx), readTimeout + additionalTimeout, TimeUnit.MILLISECONDS); - } - } - - void removeTimeout(final ChannelHandlerContext ctx) { - isTrue("Handler called from the eventLoop", ctx.channel().eventLoop().inEventLoop()); - if (timeout != null) { - timeout.cancel(false); - timeout = null; - } - } - - 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); - } - } - } - } -} From c17939f38bfb6b12ce885183c55f5a1530c81d78 Mon Sep 17 00:00:00 2001 From: Valentin Kovalenko Date: Mon, 18 Jan 2021 18:19:37 -0700 Subject: [PATCH 2/4] Suggested changes JAVA-3920 --- .../mongodb/connection/netty/NettyStream.java | 27 +++++++++---------- 1 file changed, 12 insertions(+), 15 deletions(-) diff --git a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java index 212c2e4ac37..6cb8861963d 100644 --- a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java +++ b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java @@ -102,7 +102,7 @@ * is invoked after the first operation has completed reading despite the method has not returned yet. */ final class NettyStream implements Stream { - private static final int NO_SCHEDULE_TIMEOUT = -1; + private static final byte NO_SCHEDULE_TIME = 0; private final ServerAddress address; private final SocketSettings settings; private final SslSettings sslSettings; @@ -119,13 +119,14 @@ final class NettyStream implements Stream { private PendingReader pendingReader; private Throwable pendingException; /* The fields readTimeoutTask, readTimeoutMillis are each written only in the ChannelInitializer.initChannel method - * (in addition to the write of the default value), and read only when NettyStream users read data, - * or Netty event loop handles incoming data. Since actions done by the ChannelInitializer.initChannel method + * (in addition to the write of the default value and the write by variable initializers), + * and read only when NettyStream users read data, or Netty event loop handles incoming data. + * Since actions done by the ChannelInitializer.initChannel method * are ordered (in the happens-before order) before user read actions and before event loop actions that handle incoming data, * these fields can be plain.*/ @Nullable private ReadTimeoutTask readTimeoutTask; - private long readTimeoutMillis; + private long readTimeoutMillis = NO_SCHEDULE_TIME; NettyStream(final ServerAddress address, final SocketSettings settings, final SslSettings sslSettings, final EventLoopGroup workerGroup, final Class socketChannelClass, final ByteBufAllocator allocator) { @@ -202,18 +203,14 @@ public void initChannel(final SocketChannel ch) { } int readTimeout = settings.getReadTimeout(MILLISECONDS); - final long readTimeoutMillis; - if (readTimeout > 0) { + if (readTimeout > NO_SCHEDULE_TIME) { readTimeoutMillis = readTimeout; /* We need at least one handler before (in the inbound evaluation order) the InboundBufferHandler, * so that we can fire exception events (they are inbound events) using its context and the InboundBufferHandler * receives them. SslHandler is not always present, so adding a NOOP handler.*/ pipeline.addLast(new ChannelInboundHandlerAdapter()); readTimeoutTask = new ReadTimeoutTask(pipeline.lastContext()); - } else { - readTimeoutMillis = NO_SCHEDULE_TIMEOUT; } - NettyStream.this.readTimeoutMillis = readTimeoutMillis; pipeline.addLast(new InboundBufferHandler()); } @@ -275,7 +272,7 @@ public void readAsync(final int numBytes, final AsyncCompletionHandler /** * @param numBytes Must be equal to {@link #pendingReader}{@code .numBytes} when called by a Netty channel handler. * @param handler Must be equal to {@link #pendingReader}{@code .handler} when called by a Netty channel handler. - * @param readTimeoutMillis Must be equal to {@link #NO_SCHEDULE_TIMEOUT} when called by a Netty channel handler. + * @param readTimeoutMillis Must be equal to {@link #NO_SCHEDULE_TIME} when called by a Netty channel handler. * Timeouts may be scheduled only by the public read methods. Taking into account that concurrent pending * readers are not allowed, there must not be a situation when threads attempt to schedule a timeout * before the previous one is either cancelled or completed. @@ -355,7 +352,7 @@ private void handleReadResponse(final io.netty.buffer.ByteBuf buffer, final Thro if (localPendingReader != null) { //timeouts may be scheduled only by the public read methods - readAsync(localPendingReader.numBytes, localPendingReader.handler, NO_SCHEDULE_TIMEOUT); + readAsync(localPendingReader.numBytes, localPendingReader.handler, NO_SCHEDULE_TIME); } } @@ -529,15 +526,15 @@ private static void cancel(@Nullable final Future f) { } private static long combinedTimeout(final long timeout, final int additionalTimeout) { - if (timeout == NO_SCHEDULE_TIMEOUT || additionalTimeout == NO_SCHEDULE_TIMEOUT) { - return NO_SCHEDULE_TIMEOUT; + if (timeout == NO_SCHEDULE_TIME) { + return NO_SCHEDULE_TIME; } else { return Math.addExact(timeout, additionalTimeout); } } private static ScheduledFuture scheduleReadTimeout(@Nullable final ReadTimeoutTask readTimeoutTask, final long timeoutMillis) { - if (timeoutMillis == NO_SCHEDULE_TIMEOUT) { + if (timeoutMillis == NO_SCHEDULE_TIME) { return null; } else { //assert readTimeoutTask != null : "readTimeoutTask must be initialized if read timeouts are enabled"; @@ -570,4 +567,4 @@ private ScheduledFuture schedule(final long timeoutMillis) { return ctx.executor().schedule(this, timeoutMillis, MILLISECONDS); } } -} +} \ No newline at end of file From 826ac455a18b16e47f3f53f6af3afdb392b31a95 Mon Sep 17 00:00:00 2001 From: Valentin Kovalenko Date: Tue, 19 Jan 2021 19:43:20 -0700 Subject: [PATCH 3/4] Suggested changes JAVA-3920 --- .../src/main/com/mongodb/connection/netty/NettyStream.java | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java index 6cb8861963d..d9c619902c3 100644 --- a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java +++ b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java @@ -189,7 +189,7 @@ private void initializeChannel(final AsyncCompletionHandler handler, final bootstrap.handler(new ChannelInitializer() { @Override public void initChannel(final SocketChannel ch) { - final ChannelPipeline pipeline = ch.pipeline(); + ChannelPipeline pipeline = ch.pipeline(); if (sslSettings.isEnabled()) { SSLEngine engine = getSslContext().createSSLEngine(address.getHost(), address.getPort()); engine.setUseClientMode(true); @@ -282,15 +282,10 @@ private void readAsync(final int numBytes, final AsyncCompletionHandler Throwable exceptionResult = null; synchronized (this) { exceptionResult = pendingException; - PendingReader pendingReader = this.pendingReader; if (exceptionResult == null) { if (!hasBytesAvailable(numBytes)) { if (pendingReader == null) {//called by a public read method pendingReader = new PendingReader(numBytes, handler, scheduleReadTimeout(readTimeoutTask, readTimeoutMillis)); - this.pendingReader = pendingReader; - } else {//called by a Netty channel handler - //assert pendingReader.numBytes == numBytes : "Concurrent pending readers are not allowed"; - //assert pendingReader.handler == handler : "Concurrent pending readers are not allowed"; } } else { CompositeByteBuf composite = allocator.compositeBuffer(pendingInboundBuffers.size()); From 05afdb5b9d0bcdfaa606ddc3980ba2eb8e35cde9 Mon Sep 17 00:00:00 2001 From: Valentin Kovalenko Date: Wed, 20 Jan 2021 08:36:49 -0700 Subject: [PATCH 4/4] Add the newline character sequence at the end of NettyStream.java JAVA-3920 --- .../src/main/com/mongodb/connection/netty/NettyStream.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java index d9c619902c3..42306be1335 100644 --- a/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java +++ b/driver-core/src/main/com/mongodb/connection/netty/NettyStream.java @@ -562,4 +562,4 @@ private ScheduledFuture schedule(final long timeoutMillis) { return ctx.executor().schedule(this, timeoutMillis, MILLISECONDS); } } -} \ No newline at end of file +}