Fix racy pending frame status in SimulcastEncoderAdapter. While each sub-encoder (in a StreamContext) can be assumed to process frames in order, there is the chance for a data race since the `OnEncodedImage` callback can happen asynchronously from another thread. This CL makes sure to protect the `pending_rtp_timestamps_` list using a mutex to prevent issue in such cases. A pretty complex unit test is also added, that is consistely able to reproduce this problem when executed under TSAN. Bug: webrtc:467444018 Change-Id: I595dd403558549b338ac6a67ebc87f6cb610a481 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/457620 Reviewed-by: Rasmus Brandt <brandtr@webrtc.org> Commit-Queue: Erik Språng <sprang@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47186}
diff --git a/media/BUILD.gn b/media/BUILD.gn index 9b026bf..d02dea6 100644 --- a/media/BUILD.gn +++ b/media/BUILD.gn
@@ -967,6 +967,7 @@ "../api:mock_video_bitrate_allocator_factory", "../api:mock_video_codec_factory", "../api:mock_video_decoder", + "../api:mock_video_encoder", "../api:priority", "../api:ref_count", "../api:rtc_error", @@ -986,6 +987,7 @@ "../api/crypto:options", "../api/environment", "../api/environment:environment_factory", + "../api/task_queue", "../api/test/video:function_video_factory", "../api/transport:bitrate_settings", "../api/transport:datagram_transport_interface", @@ -1037,13 +1039,17 @@ "../rtc_base:byte_order", "../rtc_base:checks", "../rtc_base:copy_on_write_buffer", + "../rtc_base:cpu_info", "../rtc_base:dscp", + "../rtc_base:macromagic", + "../rtc_base:rtc_event", "../rtc_base:safe_conversions", "../rtc_base:socket", "../rtc_base:threading", "../rtc_base:timeutils", "../rtc_base:unique_id_generator", "../rtc_base/experiments:min_video_bitrate_experiment", + "../rtc_base/synchronization:mutex", "../rtc_base/system:file_wrapper", "../system_wrappers", "../test:audio_codec_mocks",
diff --git a/media/engine/simulcast_encoder_adapter.cc b/media/engine/simulcast_encoder_adapter.cc index c25d0a5..fbed6ef 100644 --- a/media/engine/simulcast_encoder_adapter.cc +++ b/media/engine/simulcast_encoder_adapter.cc
@@ -196,9 +196,9 @@ is_keyframe_needed_(false), is_paused_(is_paused) { if (parent_) { + // If we have a parent, this encoder is not running in bypass mode and we + // should funnnel the callbacks back via this stream context. encoder_context_->encoder().RegisterEncodeCompleteCallback(this); - } else { - RTC_LOG(LS_ERROR) << "[SEA] StreamContext ctor parent is null"; } } @@ -212,6 +212,8 @@ is_keyframe_needed_(rhs.is_keyframe_needed_), is_paused_(rhs.is_paused_) { if (parent_) { + // If we have a parent, this encoder is not running in bypass mode and we + // should funnel the callbacks back via this stream context. encoder_context_->encoder().RegisterEncodeCompleteCallback(this); } } @@ -273,9 +275,13 @@ int SimulcastEncoderAdapter::StreamContext::Encode( const VideoFrame& frame, const std::vector<VideoFrameType>* frame_types) { - pending_rtp_timestamps_.push_back(frame.rtp_timestamp()); + { + MutexLock lock(&queue_mutex_); + pending_rtp_timestamps_.push_back(frame.rtp_timestamp()); + } int ret = encoder().Encode(frame, frame_types); if (ret != WEBRTC_VIDEO_CODEC_OK) { + MutexLock lock(&queue_mutex_); RTC_DCHECK_EQ(pending_rtp_timestamps_.back(), frame.rtp_timestamp()); pending_rtp_timestamps_.pop_back(); } @@ -284,15 +290,24 @@ void SimulcastEncoderAdapter::StreamContext::MaybeCullOldTimestamps( uint32_t new_rtp_timestamp) { - while (!pending_rtp_timestamps_.empty() && - pending_rtp_timestamps_.front() != new_rtp_timestamp && - IsNewerTimestamp(new_rtp_timestamp, pending_rtp_timestamps_.front())) { - parent_->OnFrameDropped(pending_rtp_timestamps_.front(), stream_idx_); - pending_rtp_timestamps_.pop_front(); + std::vector<uint32_t> dropped_timestamps; + { + MutexLock lock(&queue_mutex_); + while ( + !pending_rtp_timestamps_.empty() && + pending_rtp_timestamps_.front() != new_rtp_timestamp && + IsNewerTimestamp(new_rtp_timestamp, pending_rtp_timestamps_.front())) { + dropped_timestamps.push_back(pending_rtp_timestamps_.front()); + pending_rtp_timestamps_.pop_front(); + } + if (!pending_rtp_timestamps_.empty() && + pending_rtp_timestamps_.front() == new_rtp_timestamp) { + pending_rtp_timestamps_.pop_front(); + } } - if (!pending_rtp_timestamps_.empty() && - pending_rtp_timestamps_.front() == new_rtp_timestamp) { - pending_rtp_timestamps_.pop_front(); + // Make callback outside lock to avoid potential deadlocks. + for (uint32_t timestamp : dropped_timestamps) { + parent_->OnFrameDropped(timestamp, stream_idx_); } }
diff --git a/media/engine/simulcast_encoder_adapter.h b/media/engine/simulcast_encoder_adapter.h index 0bc6c40..99ca19d 100644 --- a/media/engine/simulcast_encoder_adapter.h +++ b/media/engine/simulcast_encoder_adapter.h
@@ -159,7 +159,8 @@ SimulcastEncoderAdapter* const parent_; std::unique_ptr<EncoderContext> encoder_context_; std::unique_ptr<FramerateController> framerate_controller_; - std::deque<uint32_t> pending_rtp_timestamps_; + mutable Mutex queue_mutex_; + std::deque<uint32_t> pending_rtp_timestamps_ RTC_GUARDED_BY(queue_mutex_); const int stream_idx_; const uint16_t width_; const uint16_t height_;
diff --git a/media/engine/simulcast_encoder_adapter_unittest.cc b/media/engine/simulcast_encoder_adapter_unittest.cc index ca0bb2b..df2894c 100644 --- a/media/engine/simulcast_encoder_adapter_unittest.cc +++ b/media/engine/simulcast_encoder_adapter_unittest.cc
@@ -10,9 +10,11 @@ #include "media/engine/simulcast_encoder_adapter.h" +#include <stddef.h> +#include <stdint.h> + #include <array> -#include <cstddef> -#include <cstdint> +#include <atomic> #include <memory> #include <optional> #include <string> @@ -21,6 +23,7 @@ #include <vector> #include "absl/container/inlined_vector.h" +#include "absl/functional/any_invocable.h" #include "api/environment/environment.h" #include "api/environment/environment_factory.h" #include "api/fec_controller_override.h" @@ -28,13 +31,17 @@ #include "api/make_ref_counted.h" #include "api/rtp_parameters.h" #include "api/scoped_refptr.h" +#include "api/task_queue/task_queue_base.h" +#include "api/task_queue/task_queue_factory.h" #include "api/test/create_simulcast_test_fixture.h" #include "api/test/mock_video_bitrate_allocator.h" #include "api/test/mock_video_decoder.h" +#include "api/test/mock_video_encoder.h" #include "api/test/simulcast_test_fixture.h" #include "api/test/video/function_video_decoder_factory.h" #include "api/test/video/function_video_encoder_factory.h" #include "api/units/data_rate.h" +#include "api/units/timestamp.h" #include "api/video/encoded_image.h" #include "api/video/i420_buffer.h" #include "api/video/video_bitrate_allocation.h" @@ -58,6 +65,10 @@ #include "modules/video_coding/utility/simulcast_rate_allocator.h" #include "modules/video_coding/utility/simulcast_test_fixture_impl.h" #include "rtc_base/checks.h" +#include "rtc_base/cpu_info.h" +#include "rtc_base/event.h" +#include "rtc_base/synchronization/mutex.h" +#include "rtc_base/thread_annotations.h" #include "test/create_test_field_trials.h" #include "test/gmock.h" #include "test/gtest.h" @@ -198,6 +209,7 @@ class MockVideoEncoderFactory : public VideoEncoderFactory { public: + explicit MockVideoEncoderFactory(const Environment& env) : env_(env) {} std::vector<SdpVideoFormat> GetSupportedFormats() const override; std::unique_ptr<VideoEncoder> Create(const Environment& env, @@ -223,10 +235,18 @@ void set_fallback_from_simulcast(std::optional<int32_t> return_value) { fallback_from_simulcast_ = return_value; } + TaskQueueBase* GetAsyncEncoderQueue() { + if (!async_encoder_queue_) { + async_encoder_queue_ = env_.task_queue_factory().CreateTaskQueue( + "AsyncEncoderQueue", TaskQueueFactory::Priority::NORMAL); + } + return async_encoder_queue_.get(); + } void DestroyVideoEncoder(VideoEncoder* encoder); private: + const Environment& env_; bool create_video_encoder_return_nullptr_ = false; int32_t init_encode_return_value_ = 0; std::optional<int32_t> fallback_from_simulcast_; @@ -236,6 +256,7 @@ std::vector<uint32_t> requested_resolution_alignments_ = {1, 1, 1}; bool supports_simulcast_ = false; std::vector<VideoEncoder::ResolutionBitrateLimits> resolution_bitrate_limits_; + std::unique_ptr<TaskQueueBase, TaskQueueDeleter> async_encoder_queue_; }; class MockVideoEncoder : public VideoEncoder { @@ -272,7 +293,11 @@ return 0; } - MOCK_METHOD(int32_t, Release, (), (override)); + MOCK_METHOD(int32_t, ReleaseMock, ()); + int32_t Release() override { + RunUntilIdle(); + return ReleaseMock(); + } void SetRates(const RateControlParameters& parameters) override { last_set_rates_ = parameters; @@ -319,6 +344,23 @@ callback_->OnEncodedImage(image, &codec_specific_info); } + void PostTask(absl::AnyInvocable<void() &&> task) { + if (is_async_) { + factory_->GetAsyncEncoderQueue()->PostTask(std::move(task)); + } else { + std::move(task)(); + } + } + + void set_is_async(bool is_async) { + if (is_async_ && !is_async) { + // Turning from asynchronous to synchronous. Wait for any pending tasks to + // complete. + RunUntilIdle(); + } + is_async_ = is_async; + } + void set_supports_native_handle(bool enabled) { supports_native_handle_ = enabled; } @@ -388,6 +430,13 @@ SdpVideoFormat video_format() const { return video_format_; } private: + void RunUntilIdle() { + // This is dangerous, only do it here for testing. + Event event; + factory_->GetAsyncEncoderQueue()->PostTask([&event] { event.Set(); }); + event.Wait(Event::kForever); + } + MockVideoEncoderFactory* const factory_; bool supports_native_handle_ = false; std::string implementation_name_ = "unknown"; @@ -405,6 +454,7 @@ std::optional<bool> is_qp_trusted_; SdpVideoFormat video_format_; std::vector<VideoEncoder::ResolutionBitrateLimits> resolution_bitrate_limits; + bool is_async_ = false; VideoCodec codec_; EncodedImageCallback* callback_; @@ -467,8 +517,9 @@ bool use_fallback_factory, const SdpVideoFormat& video_format) : env_(env), + primary_factory_(env_), fallback_factory_(use_fallback_factory - ? std::make_unique<MockVideoEncoderFactory>() + ? std::make_unique<MockVideoEncoderFactory>(env_) : nullptr), video_format_(video_format) {} @@ -508,6 +559,7 @@ env_, use_fallback_factory_, SdpVideoFormat("VP8", sdp_video_parameters_)); adapter_ = helper_->CreateMockEncoderAdapter(); + MutexLock lock(&mutex_); last_encoded_image_width_ = std::nullopt; last_encoded_image_height_ = std::nullopt; last_encoded_image_simulcast_index_ = std::nullopt; @@ -526,6 +578,7 @@ Result OnEncodedImage( const EncodedImage& encoded_image, const CodecSpecificInfo* /* codec_specific_info */) override { + MutexLock lock(&mutex_); last_encoded_image_width_ = encoded_image._encodedWidth; last_encoded_image_height_ = encoded_image._encodedHeight; last_encoded_image_simulcast_index_ = encoded_image.SimulcastIndex(); @@ -547,6 +600,7 @@ bool GetLastEncodedImageInfo(std::optional<int>* out_width, std::optional<int>* out_height, std::optional<int>* out_simulcast_index) { + MutexLock lock(&mutex_); if (!last_encoded_image_width_.has_value()) { return false; } @@ -700,13 +754,14 @@ protected: FieldTrials field_trials_ = CreateTestFieldTrials(); - Environment env_ = CreateEnvironment(field_trials_.CreateCopy()); + Environment env_ = EnvironmentFactory().Create(); std::unique_ptr<TestSimulcastEncoderAdapterFakeHelper> helper_; std::unique_ptr<VideoEncoder> adapter_; VideoCodec codec_; - std::optional<int> last_encoded_image_width_; - std::optional<int> last_encoded_image_height_; - std::optional<int> last_encoded_image_simulcast_index_; + std::optional<int> last_encoded_image_width_ RTC_GUARDED_BY(mutex_); + std::optional<int> last_encoded_image_height_ RTC_GUARDED_BY(mutex_); + std::optional<int> last_encoded_image_simulcast_index_ RTC_GUARDED_BY(mutex_); + Mutex mutex_; std::unique_ptr<VideoBitrateAllocator> rate_allocator_; bool use_fallback_factory_; CodecParameterMap sdp_video_parameters_; @@ -937,11 +992,11 @@ .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); frame_types.resize(3, VideoFrameType::kVideoFrameKey); EXPECT_EQ(0, adapter_->Encode(input_frame, &frame_types)); - EXPECT_CALL(*original_encoders[0], Release()) + EXPECT_CALL(*original_encoders[0], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); - EXPECT_CALL(*original_encoders[1], Release()) + EXPECT_CALL(*original_encoders[1], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); - EXPECT_CALL(*original_encoders[2], Release()) + EXPECT_CALL(*original_encoders[2], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); EXPECT_EQ(0, adapter_->Release()); @@ -964,9 +1019,9 @@ .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); frame_types.resize(2, VideoFrameType::kVideoFrameKey); EXPECT_EQ(0, adapter_->Encode(input_frame, &frame_types)); - EXPECT_CALL(*original_encoders[0], Release()) + EXPECT_CALL(*original_encoders[0], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); - EXPECT_CALL(*original_encoders[1], Release()) + EXPECT_CALL(*original_encoders[1], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); EXPECT_EQ(0, adapter_->Release()); @@ -986,7 +1041,7 @@ .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); frame_types.resize(1, VideoFrameType::kVideoFrameKey); EXPECT_EQ(0, adapter_->Encode(input_frame, &frame_types)); - EXPECT_CALL(*original_encoders[0], Release()) + EXPECT_CALL(*original_encoders[0], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); EXPECT_EQ(0, adapter_->Release()); @@ -1012,11 +1067,11 @@ .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); frame_types.resize(3, VideoFrameType::kVideoFrameKey); EXPECT_EQ(0, adapter_->Encode(input_frame, &frame_types)); - EXPECT_CALL(*original_encoders[0], Release()) + EXPECT_CALL(*original_encoders[0], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); - EXPECT_CALL(*new_encoders[1], Release()) + EXPECT_CALL(*new_encoders[1], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); - EXPECT_CALL(*new_encoders[2], Release()) + EXPECT_CALL(*new_encoders[2], ReleaseMock()) .WillOnce(Return(WEBRTC_VIDEO_CODEC_OK)); EXPECT_EQ(0, adapter_->Release()); } @@ -2577,14 +2632,14 @@ EXPECT_EQ(helper_->factory()->encoders()[0], prev_encoder); // Singlecast, an upper stream is active. Encoder should be recreated. - EXPECT_CALL(*prev_encoder, Release()).Times(1); + EXPECT_CALL(*prev_encoder, ReleaseMock()).Times(1); SetupCodec(/*active_streams=*/{false, true}); ASSERT_EQ(1u, helper_->factory()->encoders().size()); EXPECT_NE(helper_->factory()->encoders()[0], prev_encoder); // Singlecast, the lowest stream is active. Encoder should be recreated. prev_encoder = helper_->factory()->encoders()[0]; - EXPECT_CALL(*prev_encoder, Release()).Times(1); + EXPECT_CALL(*prev_encoder, ReleaseMock()).Times(1); SetupCodec(/*active_streams=*/{true, false}); ASSERT_EQ(1u, helper_->factory()->encoders().size()); EXPECT_NE(helper_->factory()->encoders()[0], prev_encoder); @@ -2785,5 +2840,106 @@ adapter_->Encode(input_frame, &frame_types)); } +TEST_F(TestSimulcastEncoderAdapterFake, ConcurrentEncodeAndOnEncodedImage) { + // The test setup here is quite complex, but that is needed in order to make + // sure tools like TSAN is able to reliable detect issue in the wrapper + // encoder instances used for each stream. The code below makes sure that + // there is concurrent usage of both an `Encode()` to request encoding of + // a new frame at the same time that an encoder is calling `OnEncodedImage()` + // on a separate thread. + + if (cpu_info::DetectNumberOfCores() <= 2) { + GTEST_SKIP() + << "Skipping test on low core count machine due to risk of flakiness."; + } + + SetUp(); + helper_->factory()->set_supports_simulcast(false); + SetupCodec({true, true}); + + std::vector<MockVideoEncoder*> encoders = helper_->factory()->encoders(); + ASSERT_EQ(2u, encoders.size()); + encoders[0]->set_is_async(true); + encoders[1]->set_is_async(true); + + std::atomic<bool> wait_for_second_frame{false}; + + EXPECT_CALL(*encoders[0], ReleaseMock()).WillRepeatedly(testing::Return(0)); + EXPECT_CALL(*encoders[1], ReleaseMock()).WillRepeatedly(testing::Return(0)); + + MockEncodedImageCallback mock_encoded_image_callback; + EXPECT_CALL(mock_encoded_image_callback, OnEncodedImage) + .WillRepeatedly(Return( + EncodedImageCallback::Result(EncodedImageCallback::Result::OK))); + adapter_->RegisterEncodeCompleteCallback(&mock_encoded_image_callback); + EXPECT_CALL(*encoders[0], Encode) + .WillRepeatedly([&wait_for_second_frame, encoders]( + const VideoFrame& frame, + const std::vector<VideoFrameType>* types) { + if (frame.rtp_timestamp() == 100) { + encoders[0]->PostTask([width = frame.width(), height = frame.height(), + rtp_timestamp = frame.rtp_timestamp(), + &wait_for_second_frame, + encoder = encoders[0]]() { + // Spin until the main thread is just about to encode the second + // frame. Using relaxed memory order prevents TSAN from seeing a + // happens-before edge but abort after 1000000 iterations. + constexpr int kMaxLoops = 1000000; + int loops = 0; + while (!wait_for_second_frame.load(std::memory_order_relaxed) && + ++loops <= kMaxLoops) { + } + EXPECT_LT(loops, kMaxLoops); + encoder->SendEncodedImage(width, height, rtp_timestamp, + /*simulcast_index=*/1); + }); + } + return WEBRTC_VIDEO_CODEC_OK; + }); + + EXPECT_CALL(*encoders[1], Encode) + .WillRepeatedly([&](const VideoFrame& frame, + const std::vector<VideoFrameType>* types) { + encoders[1]->SendEncodedImage(frame.width() / 2, frame.height() / 2, + frame.rtp_timestamp(), + /*simulcast_index=*/0); + return WEBRTC_VIDEO_CODEC_OK; + }); + + scoped_refptr<VideoFrameBuffer> buffer(I420Buffer::Create(1280, 720)); + + std::vector<VideoFrameType> frame_types0{VideoFrameType::kVideoFrameKey, + VideoFrameType::kVideoFrameKey}; + EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, + adapter_->Encode(VideoFrame::Builder() + .set_video_frame_buffer(buffer) + .set_rtp_timestamp(100) + .set_timestamp_ms(0) + .build(), + &frame_types0)); + + std::vector<VideoFrameType> frame_types1{VideoFrameType::kVideoFrameDelta, + VideoFrameType::kVideoFrameDelta}; + + // Signal the async thread to proceed. To maximize the chance of a collision, + // we use a relaxed atomic flag which TSAN won't treat as a happens-before + // edge. + wait_for_second_frame.store(true, std::memory_order_relaxed); + + EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, + adapter_->Encode(VideoFrame::Builder() + .set_video_frame_buffer(buffer) + .set_rtp_timestamp(90100) + .set_timestamp_ms(1000) + .build(), + &frame_types1)); + + // The callbacks should have executed via the task queue, but we don't have + // strict ordering guarantees on the main thread until we call adapter + // Release, which will indirectly call Release on the encoders, flushing + // their queues. + EXPECT_EQ(0, adapter_->Release()); +} + } // namespace test } // namespace webrtc