[FLINK-33668] Decoupling Shuffle network memory and job topology

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Done
• Fix Version/s: 1.20.0
• Component/s: Runtime / Network
• Labels: pull-request-available
• Assignee: Jiang Xin
• Reporter: Jiang Xin
• Created: 28/Nov/23 03:08
• Updated: 13/Jun/24 05:44
• Resolved: 08/Mar/24 09:46

Description

With FLINK-30469  and FLINK-31643, we have decoupled the shuffle network memory and the parallelism of tasks by limiting the number of buffers for each InputGate and ResultPartition in Hybrid Shuffle. However, when a task contains multiple ResultPartitions or many shuffle tasks are running simultaneously on the same TaskManager, "Insufficient number of network buffers" errors would still occur. This usually happens when Slot Sharing Group is enabled or a TaskManager contains multiple slots.

We want to make sure that the TaskManager does not encounter "Insufficient number of network buffers" even if there are dozens of InputGates and ResultPartitions running on the same TaskManager simultaneously. I have given this some thought, and here is my rough proposal.

1. InputGate or ResultPartition only apply for buffers from LocalBufferPool, which means that InputGate will no longer ask for exclusive buffers from NetworkBufferPool directly.
2. When creating a LocalBufferPool, we need to specify the maximum, minimum, and expected number of buffers. Whenever a new LBP is created or destroyed, a redistribution will occur, to adjust the buffer count of all LocalBufferPools using the expected value as a weight and between the minimum and maximum values. According to the test, the minimum value can be set to 4 to make the Flink Job work despite the possibility of lower performance. With this minimum value, a task with 20 shuffle edges needs only 5MB of memory to avoid "insufficient network buffer" error.
3. During runtime, InputGate and ResultPartition both calculate the number of buffers used by their internal data structures based on the pool size of their corresponding LocalBufferPool, such as the exclusive buffer queue of InputGate and BufferAccumulator of ResultPartition.

Issue Links

• causes: FLINK-35587 Job fails with "The read buffer is null in credit-based input channel" on TPC-DS 10TB benchmark
• duplicates: FLINK-33004 Decoupling topology and network memory to support complex job topologies
• links to: GitHub Pull Request #23851