An introduction to network buffers in Flink.

Questions

If you came here with the following questions, I believe this article can help answer them.

• What are network buffers, or network segments, used for? You may have encountered these terms without knowing their purpose.
• Which memory region holds network buffers, and how should its size be adjusted?
• Why does restarting with higher parallelism produce a insufficient number of network buffers error? Why do complex topologies need so many network buffers? How is the number calculated?
• How should the number of network buffers be configured?

Purpose

As the name suggests, a network buffer buffers network transfers (although it is also used for transfers that do not cross the network). Flink’s upstream and downstream tasks follow a producer-consumer model. Without buffers, producers and consumers spend substantial time waiting for downstream tasks to receive data or upstream tasks to send it. Buffers decouple the two sides, so a short fluctuation on one side should have little effect on processing on the other.

This illustrates the producer-consumer model within one process. Two tasks in the same TaskManager use this model. How is it applied when tasks run in different TaskManagers and data must cross the network?

Readers familiar with the credit-based mechanism may already understand the receiver’s buffer pool. Let us look at the sending and receiving sides together.

On the Netty server side, buffers reside only in LocalBufferPool. Subpartitions do not cache buffers or own an exclusive allocation (I discussed the resulting issues in Optimizing Backpressure in Flink’s Network Stack). On the receiving side, each channel has exclusive buffers and access to shared floating buffers. Network buffers are therefore used on both sides.

The number of buffers required by a TaskManager is the sum of the sending and receiving buffers used by all its tasks. With their locations clear, we can calculate the job’s requirements from several configuration parameters, as explained below.

Usage

Calculating Memory

The Network Segments figure shown for a TaskManager in the UI can be confusing. To see how it is calculated, consider these parameters in Flink 1.9:

Parameter names and defaults may differ between versions. Using the defaults above, a 2 GB TaskManager allocates memory as follows:

• JVM cutoff memory = 2g * containerized.heap-cutoff-ratio
• JVM heap memory = (2g - JVM cut-off 内存) * (1 - taskmanager.network.memory.fraction)
• JVM non-heap memory = 2g - JVM heap memory
• Network buffer memory = (2g - JVM cut-off 内存) * taskmanager.network.memory.fraction
• Number of network segments = Network Buffer 内存 / taskmanager.memory.segment-size

Both JVM cutoff memory and network buffer memory are direct memory. The calculated number of network segments is held in the TaskManager’s NetworkBufferPool. Tasks must request their network buffers from this pool; if allocation fails, a insufficient number of network buffers error occurs.

Now that we know the total available buffer count, how do we calculate actual demand? Consider the sending and receiving sides separately:

• The sending side shares a LocalBufferPool of size subpartitions + 1. A subpartition is a channel from this task to a downstream task. With HASH or REBALANCE connections, the subpartition count equals the number of downstream tasks.
• Receiving-side network buffers = channel count * taskmanager.network.memory.buffers-per-channel + taskmanager.network.memory.floating-buffers-per-gate. Channels receive upstream data. For HASH or REBALANCE connections, the channel count equals the number of upstream tasks.

Adding the network buffer requirements of every task on a TaskManager gives its total demand. In practice, complex topologies and differing parallelism make an exact calculation impossible without knowing scheduling details. The available approach is to estimate empirically and tune these parameters.

Monitoring Metrics

Network buffers are also useful for monitoring backpressure. Although Flink’s UI offers a backpressure monitoring tool, it is awkward to reuse in internal dashboards. Flink exposes many network buffer metrics; the most common are shown below:

outPoolUsage is the fraction of LocalBufferPool occupied by sending-side buffers. inPoolUsage is the fraction of all receiving-side buffers (exclusive + floating) in use. (inPoolUsage is not shown for local transfers.)

Putting these metrics on a dashboard makes it easy to identify where backpressure starts and locate the corresponding host and TaskManager.

An Outstanding Issue

Earlier, I mentioned the difficulty of calculating buffer requirements for complex topologies. Once the topology is built, however, each task’s requirement can be calculated from the existing parameters. After resource allocation, the task placement on each TaskManager is also known. In principle, network buffers could therefore be configured automatically from the DAG, without user intervention, instead of reporting allocation errors only after tasks start.