(spark) branch master updated: [SPARK-59707][CORE] Use ArrayDeque instead of LinkedList in TransportFrameDecoder

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 30047e1cb697 [SPARK-59707][CORE] Use ArrayDeque instead of LinkedList in TransportFrameDecoder
30047e1cb697 is described below

commit 30047e1cb69722c008aa6099a08414f15aee4af5
Author: david-mollitor-db <da...@databricks.com>
AuthorDate: Tue Sep 22 08:13:25 2026 -0700

    [SPARK-59707][CORE] Use ArrayDeque instead of LinkedList in TransportFrameDecoder
    
    ### What changes were proposed in this pull request?
    
    `TransportFrameDecoder` sits at the head of the Netty receive pipeline and reassembles
    length-prefixed frames from incoming socket reads. It keeps the input `ByteBuf`s that have
    not yet been consumed into a frame in a `buffers` field, which is accessed purely as a FIFO
    queue: appended at the tail on each `channelRead`, read/removed from the head as frames are
    decoded, plus iteration and `clear` on cleanup.
    
    This changes `buffers` from `LinkedList<ByteBuf>` to `ArrayDeque<ByteBuf>`. Every operation
    used (`add`/`addLast`, `getFirst`, `removeFirst`, enhanced-for iteration, `clear`) has
    identical FIFO semantics on `ArrayDeque`, so the swap is behavior-preserving.
    
    ### Why are the changes needed?
    
    `LinkedList` allocates a node object on every `add`. `TransportFrameDecoder.channelRead`
    runs on every inbound read of every connection (RPC and block transfer), so the old code
    produced a steady stream of short-lived list-node allocations on a hot path. `ArrayDeque`
    supports the same add-at-tail / poll-at-head access pattern with a single reused backing
    array -- the queue is typically just one buffer -- eliminating the per-read node allocation
    and improving cache locality.
    
    This mirrors SPARK-59431, which replaced `LinkedList` with `ArrayDeque` for
    `BytesToBytesMap.dataPages` for the same reason. It is an allocation micro-optimization; it
    is not expected to move throughput on its own, since the shuffle data path is dominated by
    zero-copy `FileRegion` transfers rather than framing.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    Existing `TransportFrameDecoderSuite` passes (7 tests), covering frame decoding, a length
    field split across buffers, retained frames, interception, empty/negative frame sizes, and
    consolidation -- the cases that exercise `buffers` across multiple reads. This is an
    internal data-structure change with no behavior difference, so no new tests were added.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Isaac
    
    This pull request and its description were written by Isaac.
    
    Closes #58857 from david-mollitor-db/frame-decoder-arraydeque.
    
    Authored-by: david-mollitor-db <da...@databricks.com>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../java/org/apache/spark/network/util/TransportFrameDecoder.java   | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/common/network-common/src/main/java/org/apache/spark/network/util/TransportFrameDecoder.java b/common/network-common/src/main/java/org/apache/spark/network/util/TransportFrameDecoder.java
index 6b490068507a..c920167a5c28 100644
--- a/common/network-common/src/main/java/org/apache/spark/network/util/TransportFrameDecoder.java
+++ b/common/network-common/src/main/java/org/apache/spark/network/util/TransportFrameDecoder.java
@@ -17,7 +17,7 @@
 
 package org.apache.spark.network.util;
 
-import java.util.LinkedList;
+import java.util.ArrayDeque;
 
 import com.google.common.annotations.VisibleForTesting;
 import io.netty.buffer.ByteBuf;
@@ -50,7 +50,9 @@ public class TransportFrameDecoder extends ChannelInboundHandlerAdapter {
   private static final int UNKNOWN_FRAME_SIZE = -1;
   private static final long CONSOLIDATE_THRESHOLD = 20 * 1024 * 1024;
 
-  private final LinkedList<ByteBuf> buffers = new LinkedList<>();
+  // Pending input buffers not yet consumed into a frame. ArrayDeque avoids a
+  // per-channelRead list-node allocation; used strictly FIFO.
+  private final ArrayDeque<ByteBuf> buffers = new ArrayDeque<>();
   private final ByteBuf frameLenBuf = Unpooled.buffer(LENGTH_SIZE, LENGTH_SIZE);
   private final long consolidateThreshold;
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org