Re: [PR] [client] Netty client prefer heap memory to cumulate in decoder. [fluss]

["loserwang1024 (via GitHub)" <gi...@apache.org>]

loserwang1024 commented on code in PR #2803:
URL: https://github.com/apache/fluss/pull/2803#discussion_r3076977367


##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/NettyChannelInitializer.java:
##########
@@ -46,15 +49,138 @@ protected void initChannel(SocketChannel ch) throws Exception {
         }
     }
 
-    public void addFrameDecoder(SocketChannel ch, int maxFrameLength, int initialBytesToStrip) {
-        ch.pipeline()
-                .addLast(
-                        "frameDecoder",
-                        new LengthFieldBasedFrameDecoder(
-                                maxFrameLength, 0, 4, 0, initialBytesToStrip));
+    public void addFrameDecoder(
+            SocketChannel ch, int maxFrameLength, int initialBytesToStrip, boolean preferHeap) {
+        LengthFieldBasedFrameDecoder frameDecoder =
+                new LengthFieldBasedFrameDecoder(maxFrameLength, 0, 4, 0, initialBytesToStrip);
+        if (preferHeap) {
+            frameDecoder.setCumulator(new HeapPreferringCumulator(4));
+        }
+        ch.pipeline().addLast("frameDecoder", frameDecoder);
     }
 
     public void addIdleStateHandler(SocketChannel ch) {
         ch.pipeline().addLast("idle", new IdleStateHandler(0, 0, maxIdleTimeSeconds));
     }
+
+    /**
+     * A custom {@link ByteToMessageDecoder.Cumulator} that keeps the cumulation buffer on the JVM
+     * heap to avoid off-heap memory pressure from large partial messages.
+     *
+     * <p>On native transports (epoll / kqueue), {@code EpollRecvByteAllocatorHandle.allocate()}
+     * always forces the read buffer to be direct regardless of the channel allocator. The default
+     * {@code ByteToMessageDecoder#MERGE_CUMULATOR}'s fast-path would adopt this direct buffer as
+     * the cumulation. For large partial messages, the cumulation can grow up to the maximum frame
+     * size (e.g. 32 MB), so multiple connections could consume significant off-heap memory.
+     *
+     * <p><b>Origin:</b> Derived from Netty 4.1's {@code ByteToMessageDecoder.MERGE_CUMULATOR} with
+     * the following modifications to enforce heap-backed cumulation:
+     *
+     * <ol>
+     *   <li><b>Heap-only adoption on empty cumulation:</b> When the cumulation is empty and the
+     *       incoming buffer is contiguous, Netty's MERGE_CUMULATOR adopts any incoming buffer
+     *       directly. This class only adopts heap buffers; for direct buffers, it allocates a new
+     *       heap cumulation pre-sized to the expected frame length (via {@link #allocateForFrame}).
+     *   <li><b>Direct cumulation triggers expansion:</b> Added {@code cumulation.isDirect()} as an
+     *       additional condition to trigger {@link #expandCumulation}, ensuring direct cumulations
+     *       are always replaced with heap ones during appends.
+     *   <li><b>Frame-length-aware pre-allocation:</b> New {@link #allocateForFrame} method reads
+     *       the 4-byte length field from the incoming buffer to pre-size the heap buffer to the
+     *       exact frame size, avoiding repeated re-allocations.
+     *   <li><b>Forced heap in expandCumulation:</b> The companion {@link
+     *       NettyChannelInitializer#expandCumulation} uses {@code alloc.heapBuffer()} instead of
+     *       Netty's {@code alloc.buffer()} so the replacement cumulation is always on the JVM heap.
+     * </ol>
+     */
+    static final class HeapPreferringCumulator implements ByteToMessageDecoder.Cumulator {
+
+        /** Length-field size in bytes used by the frame protocol. */
+        private final int lengthFieldSize;
+
+        public HeapPreferringCumulator(int lengthFieldSize) {
+            this.lengthFieldSize = lengthFieldSize;
+        }
+
+        @Override
+        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
+            if (cumulation == in) {
+                in.release();
+                return cumulation;
+            }
+            if (!cumulation.isReadable() && in.isContiguous()) {
+                // Adopt the incoming buffer only when it is already heap-backed. Direct buffers are
+                // always copied into a heap cumulation so downstream decoding continues to operate
+                // on heap memory.
+                if (!in.isDirect()) {
+                    cumulation.release();
+                    return in;
+                }
+                // Direct buffer — allocate a heap cumulation pre-sized to the expected frame length
+                // to avoid repeated expansion.
+                cumulation.release();
+                ByteBuf heapCumulation = allocateForFrame(alloc, in);

Review Comment:
   👌



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org