chenBright opened a new pull request, #3543: URL: https://github.com/apache/brpc/pull/3543
### What problem does this PR solve? Issue Number: resolve Problem Summary: A heap-use-after-free reported by [ASan ci](https://github.com/apache/brpc/actions/runs/34765578099/job/103745823409?pr=3541#logs) in the RTMP playback test. ```text [ RUN ] RtmpTest.fail_to_play_streams W0913 15:44:31.913728 44574 0 src/bthread/bthread.cpp:520 bthread_setconcurrency_by_tag] Fail to set concurrency by tag: 0, tag concurrency should be larger than old oncurrency. old concurrency: 9, new concurrency: 5 I0913 15:44:31.915382 44574 0 src/brpc/server.cpp:1247 StartInternal] Server[PlayingDummyService] is serving on port=8571. I0913 15:44:31.915427 44574 0 src/brpc/server.cpp:1250 StartInternal] Check out http://runnervm7g52i:8571 in web browser. I0913 15:44:31.916050 44574 0 brpc_rtmp_unittest.cpp:151 TestRtmpClientStream] TestRtmpClientStream I0913 15:44:31.919382 44578 8589936129 brpc_rtmp_unittest.cpp:249 PlayingDummyStream] PlayingDummyStream(0x60d00002d0f0) I0913 15:44:31.919835 44578 8589936129 brpc_rtmp_unittest.cpp:258 OnPlay] 127.0.0.1:51854|stream=1: Got play{stream_name=unexist_stream start=-2 duration=-1 reset=1} W0913 15:44:31.920443 44585 12884903682 src/brpc/rtmp.cpp:2060 OnStatus] 0.0.0.0:0[1] NetStream.Play.StreamNotFound: Unexist stream I0913 15:44:31.921376 44574 0 brpc_rtmp_unittest.cpp:151 TestRtmpClientStream] TestRtmpClientStream E0913 15:44:31.921964 44584 8589936641 src/brpc/rtmp.cpp:2613 OnPause] 127.0.0.1:51854[1] ignored pause(offset_ms=0) I0913 15:44:31.922031 44584 8589936641 brpc_rtmp_unittest.cpp:290 OnStop] OnStop of PlayingDummyStream=0x60d00002d0f0 I0913 15:44:31.922077 44584 8589936641 brpc_rtmp_unittest.cpp:252 ~PlayingDummyStream] ~PlayingDummyStream(0x60d00002d0f0) I0913 15:44:31.923513 44584 21474838529 brpc_rtmp_unittest.cpp:249 PlayingDummyStream] PlayingDummyStream(0x60d0000268f0) I0913 15:44:31.923800 44584 21474838529 brpc_rtmp_unittest.cpp:258 OnPlay] 127.0.0.1:51868|stream=1: Got play{stream_name=unexist_stream start=-2 duration=-1 reset=1} W0913 15:44:31.924133 44585 17179871489 src/brpc/rtmp.cpp:2060 OnStatus] 127.0.0.1:8571[1] NetStream.Play.StreamNotFound: Unexist stream E0913 15:44:31.924321 44578 17179870721 src/brpc/rtmp.cpp:2613 OnPause] 127.0.0.1:51868[1] ignored pause(offset_ms=0) I0913 15:44:31.924357 44578 17179870721 brpc_rtmp_unittest.cpp:290 OnStop] OnStop of PlayingDummyStream=0x60d0000268f0 I0913 15:44:31.924373 44578 17179870721 brpc_rtmp_unittest.cpp:252 ~PlayingDummyStream] ~PlayingDummyStream(0x60d0000268f0) #0 0x6a7f9c in brpc::RtmpStreamBase::stream_id() const /home/runner/work/brpc/brpc/test/../src/brpc/rtmp.h:581:41 #1 0x67abf2 in PlayingDummyStream::SendData() /home/runner/work/brpc/brpc/test/brpc_rtmp_unittest.cpp:336:49 #2 0x6b71b4 in PlayingDummyStream::RunSendData(void*) /home/runner/work/brpc/brpc/test/brpc_rtmp_unittest.cpp:301:37 #3 0x7fe038690592 in bthread::TaskGroup::task_runner(long) /home/runner/work/brpc/brpc/src/bthread/task_group.cpp:365:29 #4 0x7fe0385f8070 in bthread_make_fcontext (libbrpc.dbg.so+0x5f8070) 0x60d000016850 is located 32 bytes inside of 136-byte region [0x60d000016830,0x60d0000168b8) freed by thread T5 (brpc_wkr:0--109) here: #0 0x4f00ad in operator delete(void*) (/home/runner/work/brpc/brpc/test/brpc_rtmp_unittest+0x4f00ad) #1 0x6b4e71 in PlayingDummyStream::~PlayingDummyStream() /home/runner/work/brpc/brpc/test/brpc_rtmp_unittest.cpp:251:27 #2 0x7fe038a299fb in butil::SharedObject::RemoveRefManually() /home/runner/work/brpc/brpc/./src/butil/shared_object.h:52:13 #3 0x7fe038a299fb in butil::intrusive_ptr_release(butil::SharedObject*) /home/runner/work/brpc/brpc/./src/butil/shared_object.h:67:10 #4 0x7fe038a299fb in butil::intrusive_ptr<brpc::RtmpServerStream>::~intrusive_ptr() /home/runner/work/brpc/brpc/./src/butil/intrusive_ptr.hpp:89:21 #5 0x7fe038a299fb in brpc::RtmpServerStream::RunOnFailed(bthread_id_t, void*, int) /home/runner/work/brpc/brpc/src/brpc/rtmp.cpp:2699:1 #6 0x7fe03860d199 in bthread_id_error2_verbose(bthread_id_t, int, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, char const*) /home/runner/work/brpc/brpc/src/bthread/id.cpp:725:20 #7 0x7fe0386103fb in bthread::IdResetter::operator()(bthread_id_t&) const /home/runner/work/brpc/brpc/src/bthread/id.cpp:307:9 #8 0x7fe0386103fb in void bthread::ListOfABAFreeId<bthread_id_t, bthread::IdTraits>::apply<bthread::IdResetter>(bthread::IdResetter const&) /home/runner/work/brpc/brpc/./src/bthread/list_of_abafree_id.h:314:17 #9 0x7fe0386103fb in bthread_id_list_reset2(bthread_id_list_t*, int, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) /home/runner/work/brpc/brpc/src/bthread/id.cpp:745:52 #10 0x7fe0386103fb in bthread_id_list_reset2_pthreadsafe(bthread_id_list_t*, int, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, pthread_mutex_t*) /home/runner/work/brpc/brpc/src/bthread/id.cpp:770:21 #11 0x7fe038ab8929 in brpc::Socket::OnFailed(int, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) /home/runner/work/brpc/brpc/src/brpc/socket.cpp:917:5 #12 0x7fe038ab6a06 in std::enable_if<HasMemberOnFailed<brpc::Socket, int&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&>::value, void>::type brpc::VersionedRefWithId<brpc::Socket>::CallOnFailed<brpc::Socket, int&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&>(brpc::Socket*, int&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&) /home/runner/work/brpc/brpc/./src/brpc/versioned_ref_with_id.h:364:5 #13 0x7fe038ab6a06 in int brpc::VersionedRefWithId<brpc::Socket>::SetFailedImpl<int&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&>(int&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&) /home/runner/work/brpc/brpc/./src/brpc/versioned_ref_with_id.h:538:13 #14 0x7fe038ab6a06 in int brpc::VersionedRefWithId<brpc::Socket>::SetFailed<int&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&>(int&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&) /home/runner/work/brpc/brpc/./src/brpc/versioned_ref_with_id.h:519:12 #15 0x7fe038ab6a06 in brpc::Socket::SetFailed(int, char const*, ...) /home/runner/work/brpc/brpc/src/brpc/socket.cpp:1122:40 #16 0x7fe038ac3c00 in brpc::Socket::StartWrite(brpc::Socket::WriteRequest*, brpc::Socket::WriteOptions const&) /home/runner/work/brpc/brpc/src/brpc/socket.cpp:1768:13 #17 0x7fe038ac437b in brpc::Socket::Write(brpc::SocketMessagePtr<void>&, brpc::Socket::WriteOptions const*) /home/runner/work/brpc/brpc/src/brpc/socket.cpp:1697:12 #18 0x7fe038a11245 in brpc::RtmpStreamBase::SendVideoMessage(brpc::RtmpVideoMessage const&) /home/runner/work/brpc/brpc/src/brpc/rtmp.cpp:1426:23 #19 0x67aa44 in PlayingDummyStream::SendData() /home/runner/work/brpc/brpc/test/brpc_rtmp_unittest.cpp:328:9 #20 0x6b71b4 in PlayingDummyStream::RunSendData(void*) /home/runner/work/brpc/brpc/test/brpc_rtmp_unittest.cpp:301:37 #21 0x7fe038690592 in bthread::TaskGroup::task_runner(long) /home/runner/work/brpc/brpc/src/bthread/task_group.cpp:365:29 previously allocated by thread T5 (brpc_wkr:0--109) here: #0 0x4ef84d in operator new(unsigned long) (/home/runner/work/brpc/brpc/test/brpc_rtmp_unittest+0x4ef84d) #1 0x6b4361 in PlayingDummyService::NewStream(brpc::RtmpConnectRequest const&) /home/runner/work/brpc/brpc/test/brpc_rtmp_unittest.cpp:353:16 #2 0x7fe038d96ee7 in brpc::policy::RtmpChunkStream::OnCreateStream(brpc::policy::RtmpMessageHeader const&, brpc::AMFInputStream*, brpc::Socket*) /home/runner/work/brpc/brpc/src/brpc/policy/rtmp_protocol.cpp:2654:18 #3 0x7fe038d8c008 in brpc::policy::RtmpChunkStream::OnCommandMessageAMF0(brpc::policy::RtmpMessageHeader const&, butil::IOBuf*, brpc::Socket*) /home/runner/work/brpc/brpc/src/brpc/policy/rtmp_protocol.cpp:2328:12 #4 0x7fe038d7e5c2 in brpc::policy::RtmpChunkStream::OnMessage(brpc::policy::RtmpBasicHeader const&, brpc::policy::RtmpMessageHeader const&, butil::IOBuf*, brpc::Socket*) /home/runner/work/brpc/brpc/src/brpc/policy/rtmp_protocol.cpp:1872:12 #5 0x7fe038d7b2ac in brpc::policy::RtmpChunkStream::Feed(brpc::policy::RtmpBasicHeader const&, butil::IOBuf*, brpc::Socket*) /home/runner/work/brpc/brpc/src/brpc/policy/rtmp_protocol.cpp:1667:30 #6 0x7fe038d78642 in brpc::policy::RtmpContext::OnChunks(butil::IOBuf*, brpc::Socket*) /home/runner/work/brpc/brpc/src/brpc/policy/rtmp_protocol.cpp:1363:21 #7 0x7fe038d74dab in brpc::policy::RtmpContext::Feed(butil::IOBuf*, brpc::Socket*) /home/runner/work/brpc/brpc/src/brpc/policy/rtmp_protocol.cpp:1168:16 #8 0x7fe038daf3b3 in brpc::policy::ParseRtmpMessage(butil::IOBuf*, brpc::Socket*, bool, void const*) /home/runner/work/brpc/brpc/src/brpc/policy/rtmp_protocol.cpp:3497:22 #9 0x7fe03896becf in brpc::InputMessengerProcessor::CutInputMessage(brpc::InputMessenger*, unsigned long*, bool) /home/runner/work/brpc/brpc/src/brpc/input_messenger_processor.cpp:73:17 #10 0x7fe03896d442 in brpc::InputMessengerProcessor::ProcessNewMessage(long, bool, unsigned long, unsigned long, brpc::InputMessageClosure&) /home/runner/work/brpc/brpc/src/brpc/input_messenger_processor.cpp:189:26 #11 0x7fe0389680bf in brpc::InputMessenger::OnNewMessages(brpc::Socket*) /home/runner/work/brpc/brpc/src/brpc/input_messenger.cpp:152:23 #12 0x7fe038b1feca in std::function<void (brpc::Socket*)>::operator()(brpc::Socket*) const /usr/bin/../lib/gcc/x86_64-linux-gnu/12/../../../../include/c++/12/bits/std_function.h:591:9 #13 0x7fe038b1feca in brpc::Transport::OnEdge(void*) /home/runner/work/brpc/brpc/./src/brpc/transport.h:33:9 #14 0x7fe038690592 in bthread::TaskGroup::task_runner(long) /home/runner/work/brpc/brpc/src/bthread/task_group.cpp:365:29 Thread T5 (brpc_wkr:0--109) created by T0 here: #0 0x4a8e7c in pthread_create (/home/runner/work/brpc/brpc/test/brpc_rtmp_unittest+0x4a8e7c) #1 0x7fe03863088c in bthread::TaskControl::init(int) /home/runner/work/brpc/brpc/src/bthread/task_control.cpp:297:24 #2 0x7fe0385eb6c5 in bthread::get_or_new_task_control() /home/runner/work/brpc/brpc/src/bthread/bthread.cpp:112:12 #3 0x7fe0385e8d3d in bthread::start_from_non_worker(unsigned long*, bthread_attr_t const*, void* (*)(void*), void*) /home/runner/work/brpc/brpc/src/bthread/bthread.cpp:278:22 #4 0x7fe0385e8d3d in bthread_start_background /home/runner/work/brpc/brpc/src/bthread/bthread.cpp:369:12 #5 0x7fe03893c25a in brpc::GlobalInitializeOrDieImpl() /home/runner/work/brpc/brpc/src/brpc/global.cpp:698:5 #6 0x7fe037299ea7 in __pthread_once_slow nptl/./nptl/pthread_once.c:116:7 SUMMARY: AddressSanitizer: heap-use-after-free /home/runner/work/brpc/brpc/test/../src/brpc/rtmp.h:581:41 in brpc::RtmpStreamBase::stream_id() const Shadow bytes around the buggy address: 0x0c1a7fffacb0: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa 0x0c1a7fffacc0: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa 0x0c1a7fffacd0: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa 0x0c1a7ffface0: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa 0x0c1a7fffacf0: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa =>0x0c1a7fffad00: fa fa fa fa fa fa fd fd fd fd[fd]fd fd fd fd fd 0x0c1a7fffad10: fd fd fd fd fd fd fd fa fa fa fa fa fa fa fa fa 0x0c1a7fffad20: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd 0x0c1a7fffad30: fd fa fa fa fa fa fa fa fa fa fd fd fd fd fd fd 0x0c1a7fffad40: fd fd fd fd fd fd fd fd fd fd fd fd fa fa fa fa 0x0c1a7fffad50: fa fa fa fa 00 00 00 00 00 00 00 00 00 00 00 00 Shadow byte legend (one shadow byte represents 8 application bytes): Addressable: 00 Partially addressable: 01 02 03 04 05 06 07 Heap left redzone: fa Freed heap region: fd Stack left redzone: f1 Stack mid redzone: f2 Stack right redzone: f3 Stack after return: f5 Stack use after scope: f8 Global redzone: f9 Global init order: f6 Poisoned by user: f7 Container overflow: fc Array cookie: ac Intra object redzone: bb ASan internal: fe Left alloca redzone: ca Right alloca redzone: cb Shadow gap: cc ==44574==ABORTING ``` A send failure can synchronously invoke the stop callback on the sender bthread. Joining the current bthread fails, allowing the framework to release the stream while the sender continues accessing it. ### What is changed and the side effects? Changed: - Acquire a stream reference before starting the sender bthread and release it if startup fails. - Transfer ownership of that reference to the sender, keeping the stream alive until sending finishes. - Skip joining the sender when the stop callback runs on that same bthread. Side effects: - Performance effects: - Breaking backward compatibility: --- ### Check List: - Please make sure your changes are compilable. - When providing us with a new feature, it is best to add related tests. - Please follow [Contributor Covenant Code of Conduct](https://github.com/apache/brpc/blob/master/CODE_OF_CONDUCT.md). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
