Handle video layer allocation updates on the network thread VideoStreamEncoder reports layer allocation changes on the encoder queue. RtpVideoSender applied them there, under its lock, and posted the activation of the RTP modules to the network thread. The key frame of a re-enabled layer could reach RtpVideoSender before the module was active, which https://webrtc-review.googlesource.com/c/src/+/505681 worked around by enabling media ahead of the activation. Now that encoded frames are handled on the network thread, the update is posted there too, and RtpVideoSender applies it, including the module activation, right away. A frame encoded after an update is posted after it, also when a hardware encoder calls back on its own thread, so the frame is sent with the new allocation and finds its module active. The early media enable and the matching check in OnEncodedImage() are removed. Part of running the RTP send path on a single sequence, so that RTPSender and RtpVideoSender no longer need locks. Bug: webrtc:42223727 Change-Id: Icc78ca7a4580d03f47ea1cfc9514bd24fbb43fe4 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/507981 Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Reviewed-by: Erik Språng <sprang@webrtc.org> Cr-Commit-Position: refs/heads/main@{#49229}
diff --git a/call/rtp_video_sender.cc b/call/rtp_video_sender.cc index c9e4f1f..4f8655e 100644 --- a/call/rtp_video_sender.cc +++ b/call/rtp_video_sender.cc
@@ -595,15 +595,12 @@ void RtpVideoSender::SetModuleIsActive(bool sending, RtpRtcpInterface& rtp_module) { - rtp_module.SetSendingMediaStatus(sending); - // OnVideoLayersAllocationUpdated() enables media before the module is - // activated here, so the RTCP sending status tells whether the module is - // active. - if (rtp_module.Sending() == sending) { + if (rtp_module.SendingMedia() == sending) { return; } rtp_module.SetSendingStatus(sending); + rtp_module.SetSendingMediaStatus(sending); if (sending) { transport_->RegisterSendingRtpStream(rtp_module); } else { @@ -663,11 +660,7 @@ if (!rtp_streams_[simulcast_index].rtp_rtcp->OnSendingRtpFrame( encoded_image.RtpTimestamp(), encoded_image.capture_time_ms_, rtp_config_.GetStreamConfig(simulcast_index).payload_type, - encoded_image.IsKey()) && - // OnSendingRtpFrame() returns false until the module has been activated - // on the transport queue, but OnVideoLayersAllocationUpdated() enables - // media right away. Send such frames rather than drop them. - !rtp_streams_[simulcast_index].rtp_rtcp->SendingMedia()) { + encoded_image.IsKey())) { // The payload router could be active but this module isn't sending. return Result(Result::ERROR_SEND_FAILED); } @@ -730,6 +723,7 @@ void RtpVideoSender::OnVideoLayersAllocationUpdated( const VideoLayersAllocation& allocation) { + RTC_DCHECK_RUN_ON(&transport_checker_); MutexLock lock(&mutex_); if (IsActiveLocked()) { for (size_t i = 0; i < rtp_streams_.size(); ++i) { @@ -742,37 +736,16 @@ // Only send video frames on the rtp module if the encoder is configured // to send. This is to prevent stray frames to be sent after an encoder // has been reconfigured. - // Reconfiguration of the RtpRtcp modules must happen on the transport queue - // to avoid races with batch sending of packets. std::vector<bool> sending(rtp_streams_.size(), false); for (const VideoLayersAllocation::SpatialLayer& layer : allocation.active_spatial_layers) { if (layer.rtp_stream_index < static_cast<int>(sending.size())) { sending[layer.rtp_stream_index] = true; - // Enable media right away, so that the key frame the encoder produces - // for a re-enabled layer isn't dropped if it's encoded before the task - // below has run. This doesn't race with batch sending since the pacer - // has no packets for an inactive module (DeRegisterSendingRtpStream() - // prunes them, and RTPSender builds no RTX or padding packets while - // media is off). Packets of new frames are enqueued in the pacer on - // the transport queue, i.e. after the task has activated the module. - rtp_streams_[layer.rtp_stream_index].rtp_rtcp->SetSendingMediaStatus( - true); } } - transport_queue_.PostTask( - SafeTask(safety_.flag(), [this, sending = std::move(sending)] { - RTC_DCHECK_RUN_ON(&transport_checker_); - // It's possible for another task to be scheduled on the transport - // checker ahead of this call that makes the sender not active. - if (!IsActive()) { - return; - } - RTC_CHECK_EQ(sending.size(), rtp_streams_.size()); - for (size_t i = 0; i < sending.size(); ++i) { - SetModuleIsActive(sending[i], *rtp_streams_[i].rtp_rtcp); - } - })); + for (size_t i = 0; i < sending.size(); ++i) { + SetModuleIsActive(sending[i], *rtp_streams_[i].rtp_rtcp); + } } }
diff --git a/call/rtp_video_sender_interface.h b/call/rtp_video_sender_interface.h index ed95bdc..c0c289c 100644 --- a/call/rtp_video_sender_interface.h +++ b/call/rtp_video_sender_interface.h
@@ -30,9 +30,8 @@ struct FecProtectionParams; // Methods are called on the transport queue, which is the network thread and -// also runs the pacer, with these exceptions: -// - OnVideoLayersAllocationUpdated() is called on the encoder queue. -// - SetFecAllowed() may be called on any thread. +// also runs the pacer, except SetFecAllowed(), which may be called on any +// thread. class RtpVideoSenderInterface : public EncodedImageCallback, public FecControllerOverride { public:
diff --git a/call/rtp_video_sender_unittest.cc b/call/rtp_video_sender_unittest.cc index 1c6c775..5e515a2 100644 --- a/call/rtp_video_sender_unittest.cc +++ b/call/rtp_video_sender_unittest.cc
@@ -989,10 +989,9 @@ EXPECT_THAT(delta_frame_dd.frame_dependencies.frame_diffs, ElementsAre(1)); } -// The encoder produces a key frame for a re-enabled layer right away, possibly -// before the layer's RTP module has been activated on the transport queue. The -// key frame must not be dropped. -TEST(RtpVideoSenderTest, SendsKeyFrameOfReEnabledLayerBeforeModuleIsActivated) { +// The encoder produces a key frame for a re-enabled layer right away. The key +// frame must not be dropped. +TEST(RtpVideoSenderTest, SendsKeyFrameOfReEnabledLayer) { RtpVideoSenderTestFixture test({kSsrc1, kSsrc2}, {kRtxSsrc1, kRtxSsrc2}, kPayloadType, {}); test.SetSending(true); @@ -1024,8 +1023,7 @@ {.active_spatial_layers = {{.rtp_stream_index = 0}}}); test.AdvanceTime(TimeDelta::Zero()); - // Re-enable layer 1 and deliver its key frame before the task that activates - // the RTP module has run. + // Re-enable layer 1 and deliver its key frame right away. test.router()->OnVideoLayersAllocationUpdated( {.active_spatial_layers = {{.rtp_stream_index = 0}, {.rtp_stream_index = 1}}}); @@ -1799,7 +1797,7 @@ EXPECT_NE(sent_packets[0].Timestamp(), first_frame_timestamp); } -TEST(RtpVideoSenderTest, PostTaskRaceDoesNotLeadToDanglingPointer) { +TEST(RtpVideoSenderTest, AllocationUpdateWhenInactiveDoesNotReRegisterSender) { NiceMock<MockTransport> transport; NiceMock<MockRtcpIntraFrameObserver> encoder_feedback; @@ -1837,19 +1835,17 @@ EXPECT_TRUE( transport_controller.packet_router()->SsrcOfFirstSender().has_value()); - // Trigger race: - // 1. OnVideoLayersAllocationUpdated posts a task to re-register modules - // because it sees active_ is true. + // 1. Set active_ to false. + router->SetSending(false); + + // 2. An allocation update arrives afterwards, e.g. one that + // VideoSendStreamImpl posted to the transport queue before the sender was + // stopped. It must not re-register the module because active_ is now false. VideoLayersAllocation allocation; allocation.active_spatial_layers.push_back({.rtp_stream_index = 0}); router->OnVideoLayersAllocationUpdated(allocation); - // 2. Immediately set active_ to false. This happens BEFORE the posted task - // runs. - router->SetSending(false); - - // 3. Let the posted task run. With the fix, it should return early and not - // re-register the module because active_ is now false. + // 3. Let any posted tasks run. time_controller.AdvanceTime(TimeDelta::Zero()); // 4. Verify that PacketRouter does NOT have a sender.
diff --git a/video/video_send_stream_impl.cc b/video/video_send_stream_impl.cc index 562da58..be0b404 100644 --- a/video/video_send_stream_impl.cc +++ b/video/video_send_stream_impl.cc
@@ -740,9 +740,15 @@ void VideoSendStreamImpl::OnVideoLayersAllocationUpdated( VideoLayersAllocation allocation) { - // OnVideoLayersAllocationUpdated is handled on the encoder task queue in - // order to not race with OnEncodedImage callbacks. - rtp_video_sender_->OnVideoLayersAllocationUpdated(allocation); + // Called on the encoder queue. Posted to the network thread like encoded + // frames, so that frames encoded after the update are sent after it has been + // applied. + RTC_DCHECK(!network_queue_->IsCurrent()); + network_queue_->PostTask(SafeTask( + network_queue_safety_.flag(), [this, allocation = std::move(allocation)] { + RTC_DCHECK_RUN_ON(&thread_checker_); + rtp_video_sender_->OnVideoLayersAllocationUpdated(allocation); + })); } void VideoSendStreamImpl::SignalEncoderActive() {
diff --git a/video/video_send_stream_impl_unittest.cc b/video/video_send_stream_impl_unittest.cc index 6395109..b6407cf 100644 --- a/video/video_send_stream_impl_unittest.cc +++ b/video/video_send_stream_impl_unittest.cc
@@ -628,6 +628,34 @@ time_controller_.AdvanceTime(TimeDelta::Zero()); } +TEST_F(VideoSendStreamImplTest, + AppliesVideoLayersAllocationOnNetworkThreadBeforeLaterFrames) { + auto vss_impl = CreateVideoSendStreamImpl(TestVideoEncoderConfig()); + TaskQueueBase* const network_queue = time_controller_.GetMainThread(); + + Sequence s; + EXPECT_CALL(rtp_video_sender_, + OnVideoLayersAllocationUpdated(Field( + &VideoLayersAllocation::active_spatial_layers, SizeIs(1)))) + .InSequence(s) + .WillOnce([&] { EXPECT_TRUE(network_queue->IsCurrent()); }); + EXPECT_CALL(rtp_video_sender_, OnEncodedImage) + .InSequence(s) + .WillOnce(Return( + EncodedImageCallback::Result(EncodedImageCallback::Result::OK))); + + encoder_queue_->PostTask([&] { + VideoLayersAllocation allocation; + allocation.active_spatial_layers.push_back({.rtp_stream_index = 0}); + static_cast<VideoStreamEncoderInterface::EncoderSink*>(vss_impl.get()) + ->OnVideoLayersAllocationUpdated(allocation); + // A frame encoded after the update is sent after the update is applied. + static_cast<EncodedImageCallback*>(vss_impl.get()) + ->OnEncodedImage(EncodedImage(), /*codec_specific_info=*/nullptr); + }); + time_controller_.AdvanceTime(TimeDelta::Zero()); +} + TEST_F(VideoSendStreamImplTest, UpdatesObserverOnConfigurationChange) { const bool kSuspend = false; config_.suspend_below_min_bitrate = kSuspend;