[GitHub] dlg99 commented on a change in pull request #1088: ISSUE #1086 (@bug W-4146427@) Client-side backpressure in netty (Fixes: io.netty.util.internal.OutOfDirectMemoryError under continuous heavy load)

[GitBox <gi...@apache.org>]

dlg99 commented on a change in pull request #1088: ISSUE #1086 (@bug W-4146427@) Client-side backpressure in netty (Fixes: io.netty.util.internal.OutOfDirectMemoryError under continuous heavy load)
URL: https://github.com/apache/bookkeeper/pull/1088#discussion_r164893819
 
 

 ##########
 File path: bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
 ##########
 @@ -849,17 +857,84 @@ private ChannelFuture closeChannel(Channel c) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("Closing channel {}", c);
         }
-        return c.close();
+        return c.close().addListener(x -> channelChanged());
+    }
+
+    private void channelChanged() {
+        synchronized (channelMonitor) {
+            channelMonitor.notifyAll();
+        }
+    }
+
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+        channelChanged();
+        super.channelWritabilityChanged(ctx);
     }
 
     private void writeAndFlush(final Channel channel,
                                final CompletionKey key,
                                final Object request) {
-        if (channel == null) {
+        writeAndFlush(channel, key, request, -1);
+    }
+
+    private void writeAndFlush(final Channel channel,
+                           final CompletionKey key,
+                           final Object request,
+                           final long timeoutMillis) {
+        if (channel == null || !channel.isActive() || channel != this.channel) {
             errorOut(key);
             return;
         }
 
+        // channel.isWritable() check will actually respect netty's behavior configured with
+        // clientWriteBufferLowWaterMark and clientWriteBufferHighWaterMark config parameters.
+        // It is highly recommended setting them reasonably high for expected size of requests and load
+        // Otherwise client may end up doing frequent waits in this loop.
+        if (!channel.isWritable()) {
+            final long startTime = MathUtils.nowInNano();
+            final long maxSleepUntil = startTime + (timeoutMillis > 0
+                    ? TimeUnit.MILLISECONDS.toNanos(timeoutMillis)
+                    : TimeUnit.DAYS.toNanos(1));
+            synchronized (channelMonitor) {
 
 Review comment:
   I considered 1 day === infinity from request point of view. As in we are waiting until we get notified. just a way to skip "if" later

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services