This is an automated email from the ASF dual-hosted git repository.
wasphin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new e53520f4 Preserve RTMP chunk stream state on abort (#3515)
e53520f4 is described below
commit e53520f48c7b13d3bb9d0f34fa2fb80e13431043
Author: Xiaofeng Wang <[email protected]>
AuthorDate: Thu Sep 3 13:30:42 2026 +0800
Preserve RTMP chunk stream state on abort (#3515)
Reset the incomplete message when processing an RTMP Abort command instead
of deleting the chunk stream. This keeps the previous message header needed
by subsequent compressed chunk headers.
---
src/brpc/policy/rtmp_protocol.cpp | 18 +++++------
src/brpc/policy/rtmp_protocol.h | 6 ++--
test/brpc_rtmp_unittest.cpp | 63 +++++++++++++++++++++++++++++++++++----
3 files changed, 71 insertions(+), 16 deletions(-)
diff --git a/src/brpc/policy/rtmp_protocol.cpp
b/src/brpc/policy/rtmp_protocol.cpp
index d72684b1..fbc0ed0e 100644
--- a/src/brpc/policy/rtmp_protocol.cpp
+++ b/src/brpc/policy/rtmp_protocol.cpp
@@ -860,7 +860,7 @@ RtmpChunkStream* RtmpContext::GetChunkStream(uint32_t
cs_id) {
return cstream;
}
-void RtmpContext::ClearChunkStream(uint32_t cs_id) {
+void RtmpContext::AbortChunkStream(uint32_t cs_id) {
if (cs_id > RTMP_MAX_CHUNK_STREAM_ID) {
LOG(ERROR) << "Invalid chunk_stream_id=" << cs_id;
return;
@@ -879,8 +879,7 @@ void RtmpContext::ClearChunkStream(uint32_t cs_id) {
LOG(ERROR) << "chunk_stream_id=" << cs_id << " does not exist";
return;
}
- delete sub_array->ptrs[index2].exchange(
- nullptr, butil::memory_order_acquire);
+ cstream->OnAbort();
}
void RtmpContext::AllocateChunkStreamId(uint32_t* chunk_stream_id) {
@@ -1773,6 +1772,12 @@ int RtmpChunkStream::SerializeMessage(butil::IOBuf* buf,
return 0;
}
+void RtmpChunkStream::OnAbort() {
+ _r.msg_body.clear();
+ _r.left_message_length = _r.last_msg_header.message_length;
+ _r.first_chunk_of_message = true;
+}
+
static const RtmpChunkStream::MessageHandler s_msg_handlers[] = {
&RtmpChunkStream::OnSetChunkSize, // 1
&RtmpChunkStream::OnAbortMessage, // 2
@@ -1903,12 +1908,7 @@ bool RtmpChunkStream::OnAbortMessage(
RTMP_ERROR(socket, mh) << "Invalid chunk_stream_id=" << cs_id;
return false;
}
- // Do not delete the chunk stream that is currently being parsed (i.e.
- // the one running this Feed). Clearing it here would free `this' while
- // Feed() still holds and later touches it, causing a use-after-free.
- if (cs_id != _cs_id) {
- connection_context()->ClearChunkStream(cs_id);
- }
+ connection_context()->AbortChunkStream(cs_id);
return true;
}
diff --git a/src/brpc/policy/rtmp_protocol.h b/src/brpc/policy/rtmp_protocol.h
index 2ed9fd3c..c5e14ca6 100644
--- a/src/brpc/policy/rtmp_protocol.h
+++ b/src/brpc/policy/rtmp_protocol.h
@@ -306,8 +306,8 @@ public:
// Get the chunk stream by its id. The stream is created by need.
RtmpChunkStream* GetChunkStream(uint32_t cs_id);
- // Reset the chunk stream associated with the id.
- void ClearChunkStream(uint32_t cs_id);
+ // Discard the incomplete message on the chunk stream, if it exists.
+ void AbortChunkStream(uint32_t cs_id);
// Allocate/deallocate id for a chunk stream.
void AllocateChunkStreamId(uint32_t* chunk_stream_id);
@@ -424,6 +424,8 @@ public:
int SerializeMessage(butil::IOBuf* buf, const RtmpMessageHeader& mh,
butil::IOBuf* body);
+
+ void OnAbort();
bool OnMessage(
const RtmpBasicHeader& bh, const RtmpMessageHeader& mh,
diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp
index eb39f417..4315031e 100644
--- a/test/brpc_rtmp_unittest.cpp
+++ b/test/brpc_rtmp_unittest.cpp
@@ -918,11 +918,9 @@ TEST(RtmpTest, flv_reader_rejects_zero_datasize_audio_tag)
{
ASSERT_EQ(before, buf.size());
}
-// A crafted Abort message that names the chunk stream currently being parsed
-// (itself) used to make ClearChunkStream delete the RtmpChunkStream while its
-// Feed() is still running, which caused a heap-use-after-free right after
-// OnMessage() returned.
-TEST(RtmpTest, abort_message_naming_own_chunk_stream) {
+// Abort discards an incomplete message without deleting the chunk stream,
+// whether it names the stream carrying the Abort or a different stream.
+TEST(RtmpTest, abort_message_preserves_chunk_stream_state) {
int pipe_fds[2];
ASSERT_EQ(0, pipe(pipe_fds));
butil::fd_guard guard0(pipe_fds[0]); // read end, closed by this guard
@@ -973,6 +971,61 @@ TEST(RtmpTest, abort_message_naming_own_chunk_stream) {
butil::IOBuf buf2;
buf2.append(cont);
ASSERT_EQ(brpc::PARSE_OK, ctx.Feed(&buf2, sock.get()).error());
+
+ // Reduce the inbound chunk size to split four-byte control messages.
+ std::string set_chunk_size;
+ set_chunk_size.push_back((char)0x02); // fmt=0, cs_id=2
+ set_chunk_size.append(3, '\0'); // timestamp = 0
+ AppendBigEndian3Bytes(&set_chunk_size, 4);
+ set_chunk_size.push_back((char)0x01); // message_type = SetChunkSize
+ set_chunk_size.append(4, '\0'); // stream_id = 0
+ set_chunk_size.append(3, '\0');
+ set_chunk_size.push_back((char)0x02); // new chunk size = 2
+ butil::IOBuf set_chunk_size_buf;
+ set_chunk_size_buf.append(set_chunk_size);
+ ASSERT_EQ(brpc::PARSE_OK,
+ ctx.Feed(&set_chunk_size_buf, sock.get()).error());
+
+ // Start an Ack on chunk stream 3, leaving half of its body incomplete.
+ std::string partial_ack;
+ partial_ack.push_back((char)0x03); // fmt=0, cs_id=3
+ partial_ack.append(3, '\0'); // timestamp = 0
+ AppendBigEndian3Bytes(&partial_ack, 4);
+ partial_ack.push_back((char)0x03); // message_type = Ack
+ partial_ack.append(4, '\0'); // stream_id = 0
+ partial_ack.append(2, '\0'); // first two body bytes
+ butil::IOBuf partial_ack_buf;
+ partial_ack_buf.append(partial_ack);
+ ASSERT_EQ(brpc::PARSE_OK,
+ ctx.Feed(&partial_ack_buf, sock.get()).error());
+
+ // Abort chunk stream 3. The Abort itself is split over two chunks because
+ // the inbound chunk size is now two bytes.
+ std::string abort_first;
+ abort_first.push_back((char)0x02); // fmt=0, cs_id=2
+ abort_first.append(3, '\0'); // timestamp = 0
+ AppendBigEndian3Bytes(&abort_first, 4);
+ abort_first.push_back((char)0x02); // message_type = Abort
+ abort_first.append(4, '\0'); // stream_id = 0
+ abort_first.append(2, '\0'); // first two bytes of cs_id
+ butil::IOBuf abort_first_buf;
+ abort_first_buf.append(abort_first);
+ ASSERT_EQ(brpc::PARSE_OK,
+ ctx.Feed(&abort_first_buf, sock.get()).error());
+
+ const char abort_last[] = { (char)0xC2, 0, 3 }; // fmt=3, cs_id=2
+ butil::IOBuf abort_last_buf;
+ abort_last_buf.append(abort_last, sizeof(abort_last));
+ ASSERT_EQ(brpc::PARSE_OK,
+ ctx.Feed(&abort_last_buf, sock.get()).error());
+
+ // A type-3 chunk starts a new message using stream 3's previous header.
+ // Deleting the stream on Abort would lose that header and reject this.
+ const char ack_part[] = { (char)0xC3, 0, 0 }; // fmt=3, cs_id=3
+ butil::IOBuf ack_part1_buf;
+ ack_part1_buf.append(ack_part, sizeof(ack_part));
+ ASSERT_EQ(brpc::PARSE_OK,
+ ctx.Feed(&ack_part1_buf, sock.get()).error());
}
TEST(RtmpTest, rejected_create_stream_with_response_write_failure) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
