Skip to content

Commit 9f66903

Browse files
mondainmondain
authored
Merge pull request #444 from Red5/perf/rtmp-ingest-cpu
RTMP ingest: cut hot-path CPU waste and allocation churn
2 parents 038ecfc + 3b1fddb commit 9f66903

7 files changed

Lines changed: 205 additions & 23 deletions

File tree

‎RTMP_INGEST_PERF_FIXES.md‎

Lines changed: 160 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
1+
# RTMP Ingest CPU/Memory Profiling and Fixes
2+
3+
Investigation date: 2026-05-27
4+
5+
## Symptom
6+
7+
DevOps reported that a 2 vCPU instance saturated at roughly 20 concurrent RTMP
8+
publishers (plain RTMP, ~4-8 Mbps 1080p30, live fan-out to 1-5 subscribers each),
9+
which was lower throughput per core than expected.
10+
11+
## Method
12+
13+
The live system could not be profiled, so a local profiled reproduction was used:
14+
15+
- Extracted Red5 server distribution, pinned to 2 cores (`taskset -c 0,1`) with
16+
`-XX:ActiveProcessorCount=2` and the default ZGC configuration, to model a 2 vCPU box.
17+
- Load generated with ffmpeg streaming a pre-encoded 1080p30 / 6 Mbps file using
18+
`-c copy` (no encode cost), publishers and subscribers pinned to the remaining cores
19+
so they could not contaminate Red5's two cores.
20+
- 20 publishers x 3 subscribers = 80 concurrent streams (~480 Mbps relayed).
21+
- CPU and allocation profiles captured with async-profiler using the `ctimer` engine
22+
(the host has `perf_event_paranoid=4`, so perf-based events are unavailable).
23+
24+
A self-contained JMH-style micro-benchmark was also used to isolate the received-packet
25+
dispatch cost before profiling.
26+
27+
## Findings (baseline CPU profile, inclusive)
28+
29+
| Cost | % CPU | Nature |
30+
|-------------------------------------------------------|-------|--------------------------------|
31+
| Socket write syscalls (`SocketDispatcher.write0`) | ~29% | Inherent to relaying the bytes |
32+
| Socket read syscalls | ~14% | Inherent |
33+
| `String.format` inside `ChunkHeader.read` (trace log) | ~7.2% | Pure waste (fixed) |
34+
| Fan-out + per-subscriber encode | ~9% | Mostly necessary |
35+
| Received-packet dispatch (vthread + join) | ~5% | Overhead (fixed) |
36+
| ZGC | ~2.6% | Minor |
37+
38+
Allocation profile: `byte[]` 61.5% (buffer copies, inherent), `RTMP$ChannelInfo` 12.2%,
39+
and `Matcher` + `Formatter` ~3.8% (the `ChunkHeader` `String.format` again).
40+
41+
Headline conclusion: per-stream compute is small; the workload is dominated by socket
42+
I/O syscalls (~45%). The confirmed waste below is real but accounts for roughly 12% of
43+
compute, not a multiplier.
44+
45+
## Fixes in this change
46+
47+
### 1. `ChunkHeader.read` eager `String.format` in a disabled trace log
48+
49+
`common/.../net/rtmp/message/ChunkHeader.java`
50+
51+
`log.trace(...)` does not print in production (TRACE disabled), but its arguments are
52+
evaluated eagerly. `String.format("%02x", headerByte)` therefore ran on every chunk,
53+
parsing its format string via regex and allocating a `Formatter`/`Matcher`, only to
54+
discard the result. The call is now guarded with `log.isTraceEnabled()`.
55+
56+
Impact: removes ~7% of CPU and ~3.8% of allocations on the decode path.
57+
58+
### 2. Redundant per-packet virtual-thread dispatch
59+
60+
`common/.../net/rtmp/RTMPConnection.java` (`handleMessageReceived`)
61+
62+
Each received packet was wrapped in a `ReceivedMessageTask`, dispatched to a
63+
per-connection virtual-thread executor via `CompletableFuture.supplyAsync(...)`, and then
64+
immediately `join()`-ed. Because the single-threaded receiver loop blocks on the join,
65+
this provided no concurrency benefit while adding a virtual-thread spawn, a
66+
`CompletableFuture` allocation, and two context switches per packet.
67+
68+
The task now runs inline (`task.get()`) on the per-connection receiver thread. Packet
69+
ordering is preserved because that loop is single-threaded. `ReceivedMessageTask.get()`
70+
already records handler exceptions via the connection `exception` attribute, so error
71+
handling is unchanged.
72+
73+
Impact: removes the ~5% dispatch overhead and the associated context-switch and
74+
allocation churn (measured 3.6x throughput and 9x less allocation at saturation in an
75+
isolated micro-benchmark).
76+
77+
## Verification
78+
79+
Re-profiled under the identical 80-stream load after applying both fixes:
80+
81+
| Frame | Before | After |
82+
|-----------------------------------------|--------|-------|
83+
| `ChunkHeader.read` / `String.format` | 7.3% | 0.0% |
84+
| regex | 4.6% | 0.0% |
85+
| `CompletableFuture` / `supplyAsync` | 12.5% | 0.0% |
86+
| RTMP decode path (total) | 12.2% | 3.7% |
87+
88+
Both targeted paths were eliminated; nothing untouched disappeared. Overall CPU dropped
89+
single digits (the remainder is I/O-bound), giving roughly 10% more publisher headroom
90+
per core.
91+
92+
Both targeted paths were eliminated; nothing untouched disappeared.
93+
94+
## Follow-up: allocation and encode-path fixes
95+
96+
After the two fixes above, profiling showed allocation (memory cost) dominated by `byte[]`
97+
(61.5%) and `RTMP$ChannelInfo` (12.2%). Three further fixes target the encode/decode path.
98+
99+
### 3. `RTMP.getChannelInfo` allocated per call
100+
101+
`common/.../net/rtmp/codec/RTMP.java`
102+
103+
`getChannelInfo` used `channels.putIfAbsent(channelId, new ChannelInfo())`, which allocated a
104+
`ChannelInfo` on every call and discarded it whenever the channel already existed. The method
105+
is called several times per packet, so it was ~12% of all allocations. Replaced with a
106+
get-first idiom that allocates only when a channel is first seen. (A `computeIfAbsent` lambda
107+
was deliberately avoided: `ChannelInfo` is a non-static inner class, so its construction
108+
captures the enclosing instance and the lambda would itself be allocated per call.)
109+
110+
### 4. Per-chunk temporary `byte[]` in the encoder
111+
112+
`common/.../net/rtmp/codec/RTMPProtocolEncoder.java`
113+
114+
The chunk-writing loop allocated `new byte[chunkSize]` and copied through it for every chunk
115+
of every outbound message. Replaced with a direct buffer-to-buffer copy (slice the source
116+
`IoBuffer` and `put` it), eliminating the per-chunk array allocation.
117+
118+
### 5. Outbound chunk size raised 1024 -> 4096
119+
120+
`common/.../stream/consumer/ConnectionConsumer.java`
121+
122+
The outbound RTMP chunk size sent to subscribing clients was 1024 (with a "not sure of the
123+
best value" TODO). Raised to 4096 (the de-facto standard used by FFmpeg, OBS and nginx-rtmp),
124+
cutting the per-message chunk count ~4x for typical video frames - fewer chunk headers, fewer
125+
copies, less encoder work - with no client compatibility impact.
126+
127+
### 6. Per-chunk `Arrays.copyOfRange` in the decoder
128+
129+
`common/.../net/rtmp/codec/RTMPProtocolDecoder.java`
130+
131+
Chunk reassembly allocated `byte[] chunk = Arrays.copyOfRange(in.array(), ...)` for every chunk
132+
of every inbound packet, purely to transfer bytes from the input buffer into the packet buffer.
133+
Replaced with a direct buffer-to-buffer transfer (limit the source `IoBuffer` to the chunk and
134+
`buf.put(in)`), which also advances the input position so the prior explicit `skip` is removed.
135+
The per-chunk allocation is now only performed when TRACE logging is enabled.
136+
137+
## Combined verification
138+
139+
Re-profiled under the identical 80-stream load with all six fixes applied (allocation profiles
140+
are 15s windows; CPU samples are rate-normalized for comparison):
141+
142+
| Metric | Baseline | Final | Change |
143+
|--------------------------------|----------|--------|--------|
144+
| Allocation (profiler samples) | 14308 | 8236 | -42% |
145+
| CPU (rate-normalized) | 52.7/s | 46.6/s | -12% |
146+
| `RTMP$ChannelInfo` allocation | 12.2% | 0% | gone |
147+
| RTMP decode path CPU | 12.2% | ~4% | gone |
148+
| `Arrays.copyOfRange` (decode) | present | 0% | gone |
149+
| ZGC CPU | 2.6% | 1.7% | less GC|
150+
151+
The ~42% allocation reduction is the main memory win and lowers GC frequency; CPU drops ~12%,
152+
with the remainder being inherent socket I/O. The decoder change is an allocation reduction
153+
(its `copyOfRange` was only ~1-2% CPU). Remaining `byte[]` allocation is dominated by the
154+
necessary per-packet message buffers and MINA I/O buffers. RTMPChunkingTest, OriginEdgeChunkTest
155+
and RTMPExtendedTimestampTest pass, and 80 ffmpeg subscribers consumed the streams cleanly.
156+
157+
## Remaining scaling levers (not addressed)
158+
159+
- Socket read/write syscalls still dominate (~36-45% combined) and are largely inherent to
160+
per-frame low-latency relaying; coalescing writes would trade latency for fewer syscalls.

‎common/src/main/java/org/red5/server/net/rtmp/RTMPConnection.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1629,14 +1629,13 @@ public void handleMessageReceived(Packet packet) {
16291629
receivedQueueSizeUpdater.decrementAndGet(this);
16301630
// create a task to handle the packet
16311631
ReceivedMessageTask task = new ReceivedMessageTask(conn, p);
1632-
// run the task
1633-
CompletableFuture<Packet> future = CompletableFuture.supplyAsync(() -> task.get(), executor).exceptionally(throwable -> {
1634-
log.warn("Error processing received message {} state: {}", sessionId, RTMP.states[getStateCode()], throwable);
1635-
// if we have an exception, set it on the connection
1636-
conn.setAttribute("exception", throwable);
1637-
throw new CompletionException(throwable);
1638-
});
1639-
future.join();
1632+
// process the packet inline on this per-connection receiver thread. Previously this was
1633+
// dispatched to a per-connection virtual-thread executor and immediately join()-ed, which
1634+
// added a virtual-thread spawn + CompletableFuture allocation + two context switches per
1635+
// packet for no concurrency benefit (the join serialized processing anyway). Ordering is
1636+
// preserved because this loop is single-threaded. ReceivedMessageTask.get() records any
1637+
// handler exception via the connection "exception" attribute, matching prior behavior.
1638+
task.get();
16401639
}
16411640
} while (state.getState() < RTMP.STATE_ERROR); // keep processing unless we pass the error state
16421641
} catch (InterruptedException e) {

‎common/src/main/java/org/red5/server/net/rtmp/codec/RTMP.java‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -120,9 +120,18 @@ public RTMP() {
120120
* @return channel info
121121
*/
122122
private ChannelInfo getChannelInfo(int channelId) {
123-
ChannelInfo info = channels.putIfAbsent(channelId, new ChannelInfo());
123+
// get-first so the common (channel already present) path allocates nothing. This method is
124+
// called several times per packet; the previous putIfAbsent(channelId, new ChannelInfo())
125+
// allocated a ChannelInfo on every call and discarded it on a hit (~12% of all allocations).
126+
// A computeIfAbsent lambda is avoided too: ChannelInfo is a non-static inner class, so its
127+
// construction captures the enclosing instance, which would allocate a capturing lambda per call.
128+
ChannelInfo info = channels.get(channelId);
124129
if (info == null) {
125-
info = channels.get(channelId);
130+
info = new ChannelInfo();
131+
ChannelInfo existing = channels.putIfAbsent(channelId, info);
132+
if (existing != null) {
133+
info = existing;
134+
}
126135
}
127136
return info;
128137
}

‎common/src/main/java/org/red5/server/net/rtmp/codec/RTMPProtocolDecoder.java‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -328,15 +328,18 @@ public Packet decodePacket(RTMPConnection conn, RTMPDecodeState state, IoBuffer
328328
in.position(position);
329329
return null;
330330
}
331-
// get the chunk from our input
332-
byte[] chunk = Arrays.copyOfRange(in.array(), in.position(), in.position() + length);
331+
// transfer the chunk directly from the input buffer into the packet buffer. The previous
332+
// Arrays.copyOfRange allocated a byte[] for every chunk of every inbound packet, which was the
333+
// dominant byte[] allocation on the decode path under load. buf.put(in) advances in's position
334+
// by length, so the old explicit in.skip(length) is no longer needed.
335+
int inLimit = in.limit();
336+
in.limit(in.position() + length);
333337
if (isTrace) {
334-
log.trace("Read chunkSize: {}, length: {}, chunk: {}", readChunkSize, length, Hex.encodeHexString(chunk));
338+
log.trace("Read chunkSize: {}, length: {}, chunk: {}", readChunkSize, length, Hex.encodeHexString(Arrays.copyOfRange(in.array(), in.position(), in.position() + length)));
335339
}
336-
// move the position
337-
in.skip(length);
338340
// put the chunk into the packet
339-
buf.put(chunk);
341+
buf.put(in);
342+
in.limit(inLimit);
340343
if (buf.hasRemaining()) {
341344
if (isTrace) {
342345
log.trace("Packet is incomplete ({},{})", buf.remaining(), buf.limit());

‎common/src/main/java/org/red5/server/net/rtmp/codec/RTMPProtocolEncoder.java‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -185,11 +185,14 @@ public IoBuffer encodePacket(Packet packet) {
185185
do {
186186
// encode the header
187187
encodeHeader(header, lastHeader, out);
188-
// write a chunk
189-
byte[] buf = new byte[Math.min(chunkSize, data.remaining())];
190-
data.get(buf);
191-
//log.trace("Buffer: {}", Hex.encodeHexString(buf));
192-
out.put(buf);
188+
// write a chunk directly from the source buffer. The previous code allocated a
189+
// temporary byte[] per chunk and copied through it; this loop runs once per chunk
190+
// for every outbound message, so under load it was a dominant allocation source.
191+
int chunkLen = Math.min(chunkSize, data.remaining());
192+
int dataLimit = data.limit();
193+
data.limit(data.position() + chunkLen);
194+
out.put(data);
195+
data.limit(dataLimit);
193196
// move header over to last header
194197
lastHeader = header.clone();
195198
} while (data.hasRemaining());

‎common/src/main/java/org/red5/server/net/rtmp/message/ChunkHeader.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,12 @@ public static ChunkHeader read(IoBuffer in) {
147147
if (h.channelId < 0) {
148148
throw new ProtocolException("Bad channel id: " + h.channelId);
149149
}
150-
log.trace("CHUNK header byte {}, count {}, header {}, channel {}", String.format("%02x", headerByte), h.size, 0, h.channelId);
150+
if (log.isTraceEnabled()) {
151+
// String.format is evaluated eagerly as a log argument, so it must be guarded; otherwise it
152+
// runs (regex-parsing its format string and allocating a Formatter) on every chunk even when
153+
// TRACE is disabled. Profiling showed this single line at ~7% of decode CPU under load.
154+
log.trace("CHUNK header byte {}, count {}, header {}, channel {}", String.format("%02x", headerByte), h.size, 0, h.channelId);
155+
}
151156
return h;
152157
} else {
153158
// at least one byte for valid decode

‎common/src/main/java/org/red5/server/stream/consumer/ConnectionConsumer.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,10 @@ public class ConnectionConsumer implements IPushableConsumer, IPipeConnectionLis
7979
/**
8080
* Chunk size. Packets are sent chunk-by-chunk.
8181
*/
82-
private int chunkSize = 1024; //TODO: Not sure of the best value here
82+
// Outbound RTMP chunk size sent to subscribing clients. 4096 is the de-facto standard used by
83+
// FFmpeg, OBS and nginx-rtmp; raising it from 1024 cuts the per-message chunk count (and therefore
84+
// chunk-header writes and encoder work) ~4x for typical video frames with no compatibility impact.
85+
private int chunkSize = 4096;
8386

8487
/**
8588
* Whether or not the chunk size has been sent. This seems to be required for h264.

0 commit comments

Comments
 (0)