From 176611a155fecad24ef44a72efc32af9b156a51b Mon Sep 17 00:00:00 2001
From: Ry Biesemeyer <ry.biesemeyer@elastic.co>
Date: Sat, 25 Feb 2023 00:38:58 +0000
Subject: [PATCH] avoid consuming receive buffers when blocked by queue

By disabling auto-read and triggering reads when a channel becomes active
or when the previous read on a channel has completed, we avoid pulling bytes
out of our TCP receive buffer prematurely, which allows the normal mechanisms
of TCP back-pressure to work as expected.

When auto-read enabled, an input experiencing back-pressure from the
pipeline's queue would effectively avoid propagating that back-pressure
by continuing to transfer bytes from the receive buffer into effectively
unlimited direct memory buffers until it reached netty's memory allocation
limit or an OOM was reached by allocating more memory than was available.

It is likely that disabling auto-read will have some negative effect on
throughput with the default-size receive buffers, and that we will also
need to tune these or provide a way to make them user-tunable.
---
 src/main/java/org/logstash/tcp/InputLoop.java | 31 +++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/src/main/java/org/logstash/tcp/InputLoop.java b/src/main/java/org/logstash/tcp/InputLoop.java
index 8229517..1731da3 100644
--- a/src/main/java/org/logstash/tcp/InputLoop.java
+++ b/src/main/java/org/logstash/tcp/InputLoop.java
@@ -132,12 +132,17 @@ private static final class InputHandler extends ChannelInitializer<SocketChannel
         protected void initChannel(final SocketChannel channel) throws Exception {
             Decoder localCopy = decoder.copy();
 
-            // if SSL is enabled, the SSL handler must be added to the pipeline first
+            // if SSL is enabled, the SSL handler must be added to the pipeline FIRST
             if (sslContext != null) {
-                channel.pipeline().addLast(SSL_HANDLER, sslContext.newHandler(channel.alloc()));
+                channel.pipeline().addFirst(SSL_HANDLER, sslContext.newHandler(channel.alloc()));
             }
 
             channel.pipeline().addLast(new DecoderAdapter(localCopy, logger));
+
+            // disable AUTO_READ and use ThrottleReleaseHandler as LAST handler
+            channel.config().setAutoRead(false);
+            channel.pipeline().addLast(new ThrottleReleaseHandler());
+
             channel.closeFuture().addListener(new FlushOnCloseListener(localCopy));
 
             if (logger.isDebugEnabled()) {
@@ -151,6 +156,28 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
             super.exceptionCaught(ctx, cause);
         }
 
+        /**
+         * This {@link ThrottleReleaseHandler} is a handler that marks the channel eligible for
+         * reading when the channel first becomes active or has completed a read operation, and
+         * is what enables this plugin to apply TCP back-pressure when it is blocked instead of
+         * reading bytes into buffers that will vanish when we OOM.
+         *
+         * <p>It requires the channel to be configured <em>without</em> {@code AUTO_READ}</p>
+         */
+        private static final class ThrottleReleaseHandler extends ChannelInboundHandlerAdapter {
+            @Override
+            public void channelActive(final ChannelHandlerContext ctx) throws Exception {
+                super.channelActive(ctx);
+                ctx.channel().read();
+            }
+
+            @Override
+            public void channelReadComplete(final ChannelHandlerContext ctx) throws Exception {
+                super.channelReadComplete(ctx);
+                ctx.channel().read();
+            }
+        }
+
         /**
          * Listeners that flushes the the JRuby supplied {@link Decoder} when the socket is closed.
          */