Remove TaskQueueBase::Current() from Call construction This change removes implicit reliance on `TaskQueueBase::Current()` during the instantiation of `Call`, `CallStats`, `RtpTransportControllerSend`, and `TaskQueuePacedSender`. Task queue pointers are now explicitly injected through configuration structures rather than being resolved at runtime. Key modifications include: * Adding explicit `worker_task_queue` and `network_task_queue` parameters to `CallConfig` and related factories. * Replacing `sequence_checker_` with explicit thread bounds using the injected `worker_thread_` in `RtpTransportControllerSend`. * Removing the `GetCurrentTaskQueueOrThread()` fallback logic in favor of strict null checks on the injected queues. * Providing `CreateWithJoinedWorkerAndNetworkQueue` to simplify configuration where worker and network queues are identical. Bug: none Change-Id: Iad95bf32018510cace5687abe71a035c26d11726 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/478960 Reviewed-by: Harald Alvestrand <hta@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47913}
diff --git a/audio/channel_send_unittest.cc b/audio/channel_send_unittest.cc index 54a6afc..2c55a9a 100644 --- a/audio/channel_send_unittest.cc +++ b/audio/channel_send_unittest.cc
@@ -33,6 +33,7 @@ #include "api/rtp_header_extension_id.h" #include "api/rtp_headers.h" #include "api/scoped_refptr.h" +#include "api/task_queue/task_queue_base.h" #include "api/test/mock_frame_transformer.h" #include "api/test/mock_transformable_audio_frame.h" #include "api/test/rtc_error_matchers.h" @@ -85,7 +86,8 @@ {.field_trials = &field_trials_, .time = &time_controller_})), transport_controller_( RtpTransportConfig{.env = env_, - .bitrate_config = GetBitrateConfig()}) { + .bitrate_config = GetBitrateConfig(), + .worker_thread = TaskQueueBase::Current()}) { channel_ = voe::CreateChannelSend(env_, &transport_, nullptr, nullptr, crypto_options_, false, kRtcpIntervalMs, kSsrc, nullptr, &transport_controller_);
diff --git a/call/BUILD.gn b/call/BUILD.gn index 930be8e..4f85e00 100644 --- a/call/BUILD.gn +++ b/call/BUILD.gn
@@ -119,6 +119,7 @@ "../api:scoped_refptr", "../api/crypto:options", "../api/environment", + "../api/task_queue", "../api/transport:bandwidth_estimation_settings", "../api/transport:bitrate_settings", "../api/transport:network_control", @@ -641,6 +642,7 @@ "../api/audio_codecs:audio_codecs_api", "../api/environment", "../api/rtc_event_log", + "../api/task_queue", "../api/test/video:function_video_factory", "../api/units:time_delta", "../api/units:timestamp",
diff --git a/call/call.cc b/call/call.cc index 57b582a..a64e9ad 100644 --- a/call/call.cc +++ b/call/call.cc
@@ -85,7 +85,6 @@ #include "rtc_base/strings/string_builder.h" #include "rtc_base/system/no_unique_address.h" #include "rtc_base/task_utils/repeating_task.h" -#include "rtc_base/thread.h" #include "rtc_base/thread_annotations.h" #include "rtc_base/time_utils.h" #include "rtc_base/trace_event.h" @@ -150,13 +149,6 @@ return rtclog_config; } -TaskQueueBase* GetCurrentTaskQueueOrThread() { - TaskQueueBase* current = TaskQueueBase::Current(); - if (!current) - current = ThreadManager::Instance()->CurrentThread(); - return current; -} - } // namespace namespace internal { @@ -527,6 +519,8 @@ } std::unique_ptr<Call> Call::Create(CallConfig config) { + RTC_CHECK(config.worker_task_queue != nullptr); + RTC_CHECK(config.network_task_queue_ != nullptr); auto transport_send = std::make_unique<RtpTransportControllerSend>( config.ExtractTransportConfig()); @@ -706,11 +700,8 @@ Call::Call(CallConfig config, std::unique_ptr<RtpTransportControllerSendInterface> transport_send) : env_(config.env), - worker_thread_(GetCurrentTaskQueueOrThread()), - // If `network_task_queue_` was set to nullptr, network related calls - // must be made on `worker_thread_` (i.e. they're one and the same). - network_thread_(config.network_task_queue_ ? config.network_task_queue_ - : worker_thread_), + worker_thread_(config.worker_task_queue), + network_thread_(config.network_task_queue_), decode_sync_( config.decode_metronome ? std::make_unique<DecodeSynchronizer>(&env_.clock(),
diff --git a/call/call_config.cc b/call/call_config.cc index 8a076fd..086d7a6 100644 --- a/call/call_config.cc +++ b/call/call_config.cc
@@ -10,29 +10,58 @@ #include "call/call_config.h" +#include "absl/base/nullability.h" #include "api/environment/environment.h" #include "api/task_queue/task_queue_base.h" +#include "api/transport/network_types.h" #include "call/rtp_transport_config.h" +#include "rtc_base/checks.h" namespace webrtc { CallConfig::CallConfig(const Environment& env, TaskQueueBase* network_task_queue) - : env(env), network_task_queue_(network_task_queue) {} + : env(env), + network_task_queue_(network_task_queue ? network_task_queue + : TaskQueueBase::Current()), + worker_task_queue(TaskQueueBase::Current()) { + RTC_DCHECK(worker_task_queue != nullptr); + RTC_DCHECK(network_task_queue_ != nullptr); +} + +CallConfig::CallConfig(const Environment& env, + TaskQueueBase* absl_nonnull worker_task_queue, + TaskQueueBase* absl_nonnull network_task_queue) + : env(env), + network_task_queue_(network_task_queue), + worker_task_queue(worker_task_queue) { + RTC_DCHECK(worker_task_queue != nullptr); + RTC_DCHECK(network_task_queue != nullptr); +} + +CallConfig CallConfig::CreateWithJoinedWorkerAndNetworkQueue( + const Environment& env, + TaskQueueBase* absl_nonnull worker_and_network_queue) { + return CallConfig(env, worker_and_network_queue, worker_and_network_queue); +} + +CallConfig CallConfig::CreateSingleThreaded(const Environment& env) { + return CallConfig(env, TaskQueueBase::Current(), TaskQueueBase::Current()); +} RtpTransportConfig CallConfig::ExtractTransportConfig() const { - RtpTransportConfig transport_config = {.env = env}; - transport_config.bitrate_config = bitrate_config; - transport_config.network_controller_factory = - per_call_network_controller_factory - ? per_call_network_controller_factory.get() - : network_controller_factory; - transport_config.network_state_predictor_factory = - network_state_predictor_factory; - if (pacer_burst_interval.has_value()) { - transport_config.default_pacing_time_window = *pacer_burst_interval; - } - return transport_config; + return RtpTransportConfig{ + .env = env, + .bitrate_config = bitrate_config, + .network_state_predictor_factory = network_state_predictor_factory, + .network_controller_factory = + per_call_network_controller_factory + ? per_call_network_controller_factory.get() + : network_controller_factory, + .default_pacing_time_window = + pacer_burst_interval.value_or(PacerConfig::kDefaultTimeInterval), + .worker_thread = worker_task_queue, + }; } CallConfig::~CallConfig() = default;
diff --git a/call/call_config.h b/call/call_config.h index d4a2a37..aefa35f 100644 --- a/call/call_config.h +++ b/call/call_config.h
@@ -13,6 +13,7 @@ #include <memory> #include <optional> +#include "absl/base/nullability.h" #include "api/environment/environment.h" #include "api/fec_controller.h" #include "api/metronome/metronome.h" @@ -31,12 +32,26 @@ class AudioProcessing; struct CallConfig { - // If `network_task_queue` is set to nullptr, Call will assume that network - // related callbacks will be made on the same TQ as the Call instance was - // constructed on. + [[deprecated( + "Use CreateSingleThreaded or the multi-argument constructor instead.")]] explicit CallConfig(const Environment& env, TaskQueueBase* network_task_queue = nullptr); + CallConfig(const Environment& env, + TaskQueueBase* absl_nonnull worker_task_queue, + TaskQueueBase* absl_nonnull network_task_queue); + + static CallConfig CreateWithJoinedWorkerAndNetworkQueue( + const Environment& env, + TaskQueueBase* absl_nonnull worker_and_network_queue); + + // Creates a configuration for a single-threaded Call setup where signaling, + // worker, and network threads are all the same current thread. + // Note: This does not represent typical production configurations, where + // worker and network threads are usually separate background threads, but it + // is still a supported configuration (e.g. for testing or utility programs). + static CallConfig CreateSingleThreaded(const Environment& env); + // Move-only. CallConfig(CallConfig&&) = default; CallConfig& operator=(CallConfig&& other) = default; @@ -76,6 +91,7 @@ NetEqFactory* neteq_factory = nullptr; TaskQueueBase* network_task_queue_ = nullptr; + TaskQueueBase* worker_task_queue = nullptr; Metronome* decode_metronome = nullptr; Metronome* encode_metronome = nullptr;
diff --git a/call/call_unittest.cc b/call/call_unittest.cc index e024898..1c2cbed 100644 --- a/call/call_unittest.cc +++ b/call/call_unittest.cc
@@ -83,7 +83,8 @@ : make_ref_counted<NiceMock<MockAudioProcessing>>(); audio_state_config.audio_device_module = make_ref_counted<MockAudioDeviceModule>(); - CallConfig config(CreateTestEnvironment({.event_log = &log_})); + CallConfig config = CallConfig::CreateSingleThreaded( + CreateTestEnvironment({.event_log = &log_})); config.audio_state = AudioState::Create(audio_state_config); call_ = Call::Create(std::move(config)); } @@ -626,7 +627,7 @@ audio_state_config.audio_mixer = make_ref_counted<MockAudioMixer>(); audio_state_config.audio_device_module = make_ref_counted<MockAudioDeviceModule>(); - CallConfig config(env); + CallConfig config = CallConfig::CreateSingleThreaded(env); config.audio_state = AudioState::Create(audio_state_config); std::unique_ptr<Call> call(Call::Create(std::move(config)));
diff --git a/call/rtp_transport_config.h b/call/rtp_transport_config.h index 01b96b9..5252bc4 100644 --- a/call/rtp_transport_config.h +++ b/call/rtp_transport_config.h
@@ -14,6 +14,7 @@ #include "api/environment/environment.h" #include "api/network_state_predictor.h" +#include "api/task_queue/task_queue_base.h" #include "api/transport/bitrate_settings.h" #include "api/transport/network_control.h" #include "api/transport/network_types.h" @@ -37,6 +38,8 @@ // Time window used for calculating how send packets are paced. TimeDelta default_pacing_time_window = PacerConfig::kDefaultTimeInterval; + + TaskQueueBase* const worker_thread; }; } // namespace webrtc
diff --git a/call/rtp_transport_controller_send.cc b/call/rtp_transport_controller_send.cc index 2731bb8..57c098b 100644 --- a/call/rtp_transport_controller_send.cc +++ b/call/rtp_transport_controller_send.cc
@@ -98,14 +98,15 @@ RtpTransportControllerSend::RtpTransportControllerSend( const RtpTransportConfig& config) : env_(config.env), - task_queue_(TaskQueueBase::Current()), + worker_thread_(config.worker_thread), bitrate_configurator_(config.bitrate_config), pacer_started_(false), pacer_(&env_.clock(), &packet_router_, env_.field_trials(), TimeDelta::Millis(5), - 3), + 3, + worker_thread_), observer_(nullptr), controller_factory_override_(config.network_controller_factory), process_interval_(TimeDelta::PlusInfinity()), @@ -120,6 +121,7 @@ congestion_window_size_(DataSize::PlusInfinity()), is_congested_(false), retransmission_rate_limiter_(&env_.clock(), kRetransmitWindowSizeMs) { + RTC_DCHECK(worker_thread_); initial_config_.constraints = ConvertConstraints(config.bitrate_config, &env_.clock()); initial_config_.default_pacing_time_window = @@ -139,7 +141,7 @@ } RtpTransportControllerSend::~RtpTransportControllerSend() { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); RTC_DCHECK(video_rtp_senders_.empty()); pacer_queue_update_task_.Stop(); controller_task_.Stop(); @@ -155,9 +157,9 @@ std::unique_ptr<FecController> fec_controller, const RtpSenderFrameEncryptionConfig& frame_encryption_config, scoped_refptr<FrameTransformerInterface> frame_transformer) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); video_rtp_senders_.push_back(std::make_unique<RtpVideoSender>( - env_, task_queue_, suspended_ssrcs, states, rtp_config, + env_, worker_thread_, suspended_ssrcs, states, rtp_config, rtcp_report_interval_ms, send_transport, observers, // TODO(holmer): Remove this circular dependency by injecting // the parts of RtpTransportControllerSendInterface that are really used. @@ -169,7 +171,7 @@ void RtpTransportControllerSend::DestroyRtpVideoSender( RtpVideoSenderInterface* rtp_video_sender) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); std::vector<std::unique_ptr<RtpVideoSenderInterface>>::iterator it = video_rtp_senders_.end(); for (it = video_rtp_senders_.begin(); it != video_rtp_senders_.end(); ++it) { @@ -183,7 +185,7 @@ void RtpTransportControllerSend::RegisterSendingRtpStream( RtpRtcpInterface& rtp_module) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); // Allow pacer to send packets using this module. packet_router_.AddSendRtpModule(&rtp_module, /*remb_candidate=*/true); @@ -194,7 +196,7 @@ void RtpTransportControllerSend::DeRegisterSendingRtpStream( RtpRtcpInterface& rtp_module) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); // Disabling media, remove from packet router map to reduce size and // prevent any stray packets in the pacer from asynchronously arriving // to a disabled module. @@ -253,14 +255,14 @@ void RtpTransportControllerSend::SetAllocatedSendBitrateLimits( BitrateAllocationLimits limits) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); streams_config_.min_total_allocated_bitrate = limits.min_allocatable_rate; streams_config_.max_padding_rate = limits.max_padding_rate; streams_config_.max_total_allocated_bitrate = limits.max_allocatable_rate; UpdateStreamsConfig(); } void RtpTransportControllerSend::SetPacingFactor(float pacing_factor) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); // TODO: bugs.webrtc.org/447037083 - Remove or update usage of SetPacingFactor // if RFC 8888 is enabled. With RFC 8888 feedback, this method is not // invoked. Goog CC sets a sensible pacing factor by itself. @@ -277,7 +279,7 @@ void RtpTransportControllerSend::ReconfigureBandwidthEstimation( const BandwidthEstimationSettings& settings) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); bwe_settings_ = settings; streams_config_.enable_repeated_initial_probing = @@ -303,7 +305,7 @@ void RtpTransportControllerSend::RegisterTargetTransferRateObserver( TargetTransferRateObserver* observer) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); RTC_DCHECK(observer_ == nullptr); observer_ = observer; observer_->OnStartRateUpdate(*initial_config_.constraints.starting_rate); @@ -331,7 +333,7 @@ void RtpTransportControllerSend::OnNetworkRouteChanged( absl::string_view transport_name, const NetworkRoute& network_route) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); // Check if the network route is connected. if (!network_route.connected) { // TODO(honghaiz): Perhaps handle this in SignalChannelNetworkState and @@ -420,7 +422,7 @@ } void RtpTransportControllerSend::OnNetworkAvailability(bool network_available) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); RTC_LOG(LS_VERBOSE) << "SignalNetworkState " << (network_available ? "Up" : "Down"); network_available_ = network_available; @@ -451,7 +453,7 @@ return pacer_.FirstSentPacketTime(); } void RtpTransportControllerSend::EnablePeriodicAlrProbing(bool enable) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); streams_config_.requests_alr_probing = enable; UpdateStreamsConfig(); @@ -460,23 +462,23 @@ const SentPacketInfo& sent_packet) { // Normally called on the network thread! // TODO(crbug.com/1373439): Clarify other thread contexts calling in, - // and simplify task posting logic when the combined network/worker project - // launches. - if (TaskQueueBase::Current() != task_queue_) { - task_queue_->PostTask(SafeTask(safety_.flag(), [this, sent_packet]() { - RTC_DCHECK_RUN_ON(&sequence_checker_); + // and simplify task posting logic now that the combined network/worker + // project has launched. + if (TaskQueueBase::Current() != worker_thread_) { + worker_thread_->PostTask(SafeTask(safety_.flag(), [this, sent_packet]() { + RTC_DCHECK_RUN_ON(worker_thread_); ProcessSentPacket(sent_packet); })); return; } - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); ProcessSentPacket(sent_packet); } void RtpTransportControllerSend::ProcessSentPacket( const SentPacketInfo& sent_packet) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); std::optional<SentPacket> packet_msg = transport_feedback_adapter_.ProcessSentPacket(sent_packet); if (!packet_msg) @@ -491,10 +493,10 @@ ProcessSentPacketUpdates(std::move(control_update)); } -// RTC_RUN_ON(task_queue_) +// RTC_RUN_ON(worker_thread_) void RtpTransportControllerSend::ProcessSentPacketUpdates( NetworkControlUpdate updates) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); // Only update outstanding data if: // 1. Packet feedback is used. // 2. The packet has not yet received an acknowledgement. @@ -507,14 +509,14 @@ void RtpTransportControllerSend::OnReceivedPacket( const ReceivedPacket& packet_msg) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (controller_) PostUpdates(controller_->OnReceivedPacket(packet_msg)); } void RtpTransportControllerSend::UpdateBitrateConstraints( const BitrateConstraints& updated) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); TargetRateConstraints msg = ConvertConstraints(updated, &env_.clock()); if (controller_) { PostUpdates(controller_->OnTargetRateConstraints(msg)); @@ -525,7 +527,7 @@ void RtpTransportControllerSend::SetSdpBitrateParameters( const BitrateConstraints& constraints) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); std::optional<BitrateConstraints> updated = bitrate_configurator_.UpdateWithSdpParameters(constraints); if (updated.has_value()) { @@ -539,7 +541,7 @@ void RtpTransportControllerSend::SetClientBitratePreferences( const BitrateSettings& preferences) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); std::optional<BitrateConstraints> updated = bitrate_configurator_.UpdateWithClientPreferences(preferences); if (updated.has_value()) { @@ -553,7 +555,7 @@ void RtpTransportControllerSend::OnTransportOverheadChanged( size_t transport_overhead_bytes_per_packet) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (transport_overhead_bytes_per_packet >= kMaxOverheadBytes) { RTC_LOG(LS_ERROR) << "Transport overhead exceeds " << kMaxOverheadBytes; return; @@ -580,7 +582,7 @@ } void RtpTransportControllerSend::EnsureStarted() { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (!pacer_started_) { pacer_started_ = true; pacer_.EnsureStarted(); @@ -590,7 +592,7 @@ void RtpTransportControllerSend::OnReceiverEstimatedMaxBitrate( Timestamp receive_time, DataRate bitrate) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); RemoteBitrateReport msg; msg.receive_time = receive_time; msg.bandwidth = bitrate; @@ -600,7 +602,7 @@ void RtpTransportControllerSend::OnRttUpdate(Timestamp receive_time, TimeDelta rtt) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); RoundTripTimeUpdate report; report.receive_time = receive_time; report.round_trip_time = rtt.RoundTo(TimeDelta::Millis(1)); @@ -612,7 +614,7 @@ void RtpTransportControllerSend::NotifyBweOfPacedSentPacket( const RtpPacketToSend& packet, const PacedPacketInfo& pacing_info) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (!packet.transport_sequence_number()) { return; @@ -628,7 +630,7 @@ void RtpTransportControllerSend::SetPreferredRtcpCcAckType( RtcpFeedbackType preferred_rtcp_cc_ack_type) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); RTC_DCHECK(preferred_rtcp_cc_ack_type == RtcpFeedbackType::CCFB || preferred_rtcp_cc_ack_type == RtcpFeedbackType::TRANSPORT_CC); if (preferred_rtcp_cc_ack_type == RtcpFeedbackType::CCFB) { @@ -654,7 +656,7 @@ std::optional<int> RtpTransportControllerSend::ReceivedCongestionControlFeedbackCount() const { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (!rfc_8888_feedback_negotiated_) { return std::nullopt; } @@ -663,13 +665,13 @@ flat_map<uint32_t, ReceivedCongestionControlFeedbackStats> RtpTransportControllerSend::GetCongestionControlFeedbackStatsPerSsrc() const { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); return received_ccfb_stats_; } std::optional<int> RtpTransportControllerSend::ReceivedTransportCcFeedbackCount() const { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (rfc_8888_feedback_negotiated_) { return std::nullopt; } @@ -679,7 +681,7 @@ void RtpTransportControllerSend::OnTransportFeedback( Timestamp receive_time, const rtcp::TransportFeedback& feedback) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); ++transport_cc_feedback_count_; std::optional<TransportPacketsFeedback> feedback_msg = transport_feedback_adapter_.ProcessTransportFeedback(feedback, @@ -692,7 +694,7 @@ void RtpTransportControllerSend::OnCongestionControlFeedback( Timestamp receive_time, const rtcp::CongestionControlFeedback& feedback) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); ++feedback_count_; std::optional<TransportPacketsFeedback> feedback_msg = transport_feedback_adapter_.ProcessCongestionControlFeedback( @@ -783,7 +785,7 @@ void RtpTransportControllerSend::OnRemoteNetworkEstimate( NetworkStateEstimate estimate) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); estimate.update_time = env_.clock().CurrentTime(); if (controller_) PostUpdates(controller_->OnNetworkStateEstimate(estimate)); @@ -836,11 +838,11 @@ } void RtpTransportControllerSend::StartProcessPeriodicTasks() { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (!pacer_queue_update_task_.Running()) { pacer_queue_update_task_ = RepeatingTaskHandle::DelayedStart( - task_queue_, kPacerQueueUpdateInterval, [this]() { - RTC_DCHECK_RUN_ON(&sequence_checker_); + worker_thread_, kPacerQueueUpdateInterval, [this]() { + RTC_DCHECK_RUN_ON(worker_thread_); TimeDelta expected_queue_time = pacer_.ExpectedQueueTime(); control_handler_->SetPacerQueue(expected_queue_time); UpdateControlState(); @@ -850,8 +852,8 @@ controller_task_.Stop(); if (process_interval_.IsFinite()) { controller_task_ = RepeatingTaskHandle::DelayedStart( - task_queue_, process_interval_, [this]() { - RTC_DCHECK_RUN_ON(&sequence_checker_); + worker_thread_, process_interval_, [this]() { + RTC_DCHECK_RUN_ON(worker_thread_); UpdateControllerWithTimeInterval(); return process_interval_; }); @@ -893,7 +895,7 @@ void RtpTransportControllerSend::OnReport( Timestamp receive_time, std::span<const ReportBlockData> report_blocks) { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); if (report_blocks.empty()) return;
diff --git a/call/rtp_transport_controller_send.h b/call/rtp_transport_controller_send.h index 5a45ec2..eafc431 100644 --- a/call/rtp_transport_controller_send.h +++ b/call/rtp_transport_controller_send.h
@@ -140,7 +140,7 @@ void OnRemoteNetworkEstimate(NetworkStateEstimate estimate) override; NetworkControllerInterface* GetNetworkController() override { - RTC_DCHECK_RUN_ON(&sequence_checker_); + RTC_DCHECK_RUN_ON(worker_thread_); return controller_.get(); } @@ -158,96 +158,94 @@ void NotifyBweOfSentPacketForTesting(const RtpPacketToSend& rtp_packet); private: - void MaybeCreateControllers() RTC_RUN_ON(sequence_checker_); + void MaybeCreateControllers() RTC_RUN_ON(worker_thread_); void HandleTransportPacketsFeedback(const TransportPacketsFeedback& feedback) - RTC_RUN_ON(sequence_checker_); + RTC_RUN_ON(worker_thread_); void ComputeStatsFromCongestionControlFeedback( - const TransportPacketsFeedback& feedback) RTC_RUN_ON(sequence_checker_); - void UpdateNetworkAvailability() RTC_RUN_ON(sequence_checker_); + const TransportPacketsFeedback& feedback) RTC_RUN_ON(worker_thread_); + void UpdateNetworkAvailability() RTC_RUN_ON(worker_thread_); void UpdateInitialConstraints(TargetRateConstraints new_contraints) - RTC_RUN_ON(sequence_checker_); + RTC_RUN_ON(worker_thread_); - void StartProcessPeriodicTasks() RTC_RUN_ON(sequence_checker_); - void UpdateControllerWithTimeInterval() RTC_RUN_ON(sequence_checker_); + void StartProcessPeriodicTasks() RTC_RUN_ON(worker_thread_); + void UpdateControllerWithTimeInterval() RTC_RUN_ON(worker_thread_); bool IsRelevantRouteChange(const NetworkRoute& old_route, const NetworkRoute& new_route) const; void UpdateBitrateConstraints(const BitrateConstraints& updated); - void UpdateStreamsConfig() RTC_RUN_ON(sequence_checker_); - void PostUpdates(NetworkControlUpdate update) RTC_RUN_ON(sequence_checker_); - void UpdateControlState() RTC_RUN_ON(sequence_checker_); - void UpdateCongestedState() RTC_RUN_ON(sequence_checker_); + void UpdateStreamsConfig() RTC_RUN_ON(worker_thread_); + void PostUpdates(NetworkControlUpdate update) RTC_RUN_ON(worker_thread_); + void UpdateControlState() RTC_RUN_ON(worker_thread_); + void UpdateCongestedState() RTC_RUN_ON(worker_thread_); std::optional<bool> GetCongestedStateUpdate() const - RTC_RUN_ON(sequence_checker_); + RTC_RUN_ON(worker_thread_); // Called by packet router just before packet is sent to the RTP modules. void NotifyBweOfPacedSentPacket(const RtpPacketToSend& packet, const PacedPacketInfo& pacing_info); void ProcessSentPacket(const SentPacketInfo& sent_packet) - RTC_RUN_ON(sequence_checker_); + RTC_RUN_ON(worker_thread_); void ProcessSentPacketUpdates(NetworkControlUpdate updates) - RTC_RUN_ON(sequence_checker_); + RTC_RUN_ON(worker_thread_); const Environment env_; - SequenceChecker sequence_checker_; - TaskQueueBase* task_queue_; + TaskQueueBase* const worker_thread_; PacketRouter packet_router_; std::vector<std::unique_ptr<RtpVideoSenderInterface>> video_rtp_senders_ - RTC_GUARDED_BY(&sequence_checker_); + RTC_GUARDED_BY(worker_thread_); RtpBitrateConfigurator bitrate_configurator_; std::map<std::string, NetworkRoute> network_routes_ - RTC_GUARDED_BY(sequence_checker_); - BandwidthEstimationSettings bwe_settings_ RTC_GUARDED_BY(sequence_checker_); - bool pacer_started_ RTC_GUARDED_BY(sequence_checker_); + RTC_GUARDED_BY(worker_thread_); + BandwidthEstimationSettings bwe_settings_ RTC_GUARDED_BY(worker_thread_); + bool pacer_started_ RTC_GUARDED_BY(worker_thread_); TaskQueuePacedSender pacer_; - TargetTransferRateObserver* observer_ RTC_GUARDED_BY(sequence_checker_); + TargetTransferRateObserver* observer_ RTC_GUARDED_BY(worker_thread_); TransportFeedbackDemuxer feedback_demuxer_; TransportFeedbackAdapter transport_feedback_adapter_ - RTC_GUARDED_BY(sequence_checker_); + RTC_GUARDED_BY(worker_thread_); NetworkControllerFactoryInterface* const controller_factory_override_ - RTC_PT_GUARDED_BY(sequence_checker_); + RTC_PT_GUARDED_BY(worker_thread_); std::unique_ptr<CongestionControlHandler> control_handler_ - RTC_GUARDED_BY(sequence_checker_) RTC_PT_GUARDED_BY(sequence_checker_); + RTC_GUARDED_BY(worker_thread_) RTC_PT_GUARDED_BY(worker_thread_); std::unique_ptr<NetworkControllerInterface> controller_ - RTC_GUARDED_BY(sequence_checker_) RTC_PT_GUARDED_BY(sequence_checker_); + RTC_GUARDED_BY(worker_thread_) RTC_PT_GUARDED_BY(worker_thread_); - TimeDelta process_interval_ RTC_GUARDED_BY(sequence_checker_); + TimeDelta process_interval_ RTC_GUARDED_BY(worker_thread_); struct LossReport { uint32_t extended_highest_sequence_number = 0; int cumulative_lost = 0; }; std::map<uint32_t, LossReport> last_report_blocks_ - RTC_GUARDED_BY(sequence_checker_); + RTC_GUARDED_BY(worker_thread_); flat_map<uint32_t, ReceivedCongestionControlFeedbackStats> - received_ccfb_stats_ RTC_GUARDED_BY(sequence_checker_); - Timestamp last_report_block_time_ RTC_GUARDED_BY(sequence_checker_); + received_ccfb_stats_ RTC_GUARDED_BY(worker_thread_); + Timestamp last_report_block_time_ RTC_GUARDED_BY(worker_thread_); - NetworkControllerConfig initial_config_ RTC_GUARDED_BY(sequence_checker_); - StreamsConfig streams_config_ RTC_GUARDED_BY(sequence_checker_); + NetworkControllerConfig initial_config_ RTC_GUARDED_BY(worker_thread_); + StreamsConfig streams_config_ RTC_GUARDED_BY(worker_thread_); const bool add_pacing_to_cwin_; const bool reset_bwe_on_adapter_id_change_; - size_t transport_overhead_bytes_per_packet_ RTC_GUARDED_BY(sequence_checker_); - bool network_available_ RTC_GUARDED_BY(sequence_checker_); - RepeatingTaskHandle pacer_queue_update_task_ - RTC_GUARDED_BY(sequence_checker_); - RepeatingTaskHandle controller_task_ RTC_GUARDED_BY(sequence_checker_); + size_t transport_overhead_bytes_per_packet_ RTC_GUARDED_BY(worker_thread_); + bool network_available_ RTC_GUARDED_BY(worker_thread_); + RepeatingTaskHandle pacer_queue_update_task_ RTC_GUARDED_BY(worker_thread_); + RepeatingTaskHandle controller_task_ RTC_GUARDED_BY(worker_thread_); - DataSize congestion_window_size_ RTC_GUARDED_BY(sequence_checker_); - bool is_congested_ RTC_GUARDED_BY(sequence_checker_); + DataSize congestion_window_size_ RTC_GUARDED_BY(worker_thread_); + bool is_congested_ RTC_GUARDED_BY(worker_thread_); bool rfc_8888_feedback_negotiated_ = false; bool sending_packets_as_ect1_ = false; // Count of feedback messages received. - int feedback_count_ RTC_GUARDED_BY(sequence_checker_) = 0; - int transport_cc_feedback_count_ RTC_GUARDED_BY(sequence_checker_) = 0; + int feedback_count_ RTC_GUARDED_BY(worker_thread_) = 0; + int transport_cc_feedback_count_ RTC_GUARDED_BY(worker_thread_) = 0; // Protected by internal locks. RateLimiter retransmission_rate_limiter_;
diff --git a/call/rtp_transport_controller_send_unittest.cc b/call/rtp_transport_controller_send_unittest.cc index 5bf29ea7..d6ae7d3 100644 --- a/call/rtp_transport_controller_send_unittest.cc +++ b/call/rtp_transport_controller_send_unittest.cc
@@ -103,7 +103,9 @@ TEST(RtpTransportControllerSendTest, IgnoresFeedbackForReportedReceivedPacketThatWereNotSent) { test::RunLoop main_thread; - RtpTransportControllerSend transport({.env = CreateTestEnvironment()}); + RtpTransportControllerSend transport( + {.env = CreateTestEnvironment(), + .worker_thread = main_thread.task_queue()}); transport.SetPreferredRtcpCcAckType(RtcpFeedbackType::CCFB); PacketSender sender(transport); sender.SimulateSentPackets({.ssrc = 123, @@ -132,7 +134,9 @@ constexpr uint32_t kSsrc1 = 1'000; constexpr uint32_t kSsrc2 = 2'000; test::RunLoop main_thread; - RtpTransportControllerSend transport({.env = CreateTestEnvironment()}); + RtpTransportControllerSend transport( + {.env = CreateTestEnvironment(), + .worker_thread = main_thread.task_queue()}); transport.SetPreferredRtcpCcAckType(RtcpFeedbackType::CCFB); PacketSender sender(transport); @@ -177,7 +181,9 @@ TEST(RtpTransportControllerSendTest, CalculatesNumberOfBleachedPackets) { test::RunLoop main_thread; - RtpTransportControllerSend transport({.env = CreateTestEnvironment()}); + RtpTransportControllerSend transport( + {.env = CreateTestEnvironment(), + .worker_thread = main_thread.task_queue()}); transport.SetPreferredRtcpCcAckType(RtcpFeedbackType::CCFB); PacketSender sender(transport); @@ -212,7 +218,9 @@ TEST(RtpTransportControllerSendTest, AccumulatesNumberOfReportedLostAndRecoveredPackets) { test::RunLoop main_thread; - RtpTransportControllerSend transport({.env = CreateTestEnvironment()}); + RtpTransportControllerSend transport( + {.env = CreateTestEnvironment(), + .worker_thread = main_thread.task_queue()}); transport.SetPreferredRtcpCcAckType(RtcpFeedbackType::CCFB); PacketSender sender(transport); @@ -259,7 +267,9 @@ TEST(RtpTransportControllerSendTest, DoesNotCountGapsInSequenceNumberBetweenReportsAsLoss) { test::RunLoop main_thread; - RtpTransportControllerSend transport({.env = CreateTestEnvironment()}); + RtpTransportControllerSend transport( + {.env = CreateTestEnvironment(), + .worker_thread = main_thread.task_queue()}); transport.SetPreferredRtcpCcAckType(RtcpFeedbackType::CCFB); PacketSender sender(transport);
diff --git a/call/rtp_video_sender_unittest.cc b/call/rtp_video_sender_unittest.cc index e7a6e85..9676ad8 100644 --- a/call/rtp_video_sender_unittest.cc +++ b/call/rtp_video_sender_unittest.cc
@@ -190,8 +190,10 @@ payload_type, payload_types)), bitrate_config_(GetBitrateConfig()), - transport_controller_( - RtpTransportConfig{.env = env_, .bitrate_config = bitrate_config_}), + transport_controller_(RtpTransportConfig{ + .env = env_, + .bitrate_config = bitrate_config_, + .worker_thread = time_controller_.GetMainThread()}), stats_proxy_(time_controller_.GetClock(), config_, VideoEncoderConfig::ContentType::kRealtimeVideo, @@ -1625,8 +1627,10 @@ VideoEncoderConfig::ContentType::kRealtimeVideo, env.field_trials()); BitrateConstraints bitrate_config = GetBitrateConfig(); - RtpTransportConfig transport_config{.env = env, - .bitrate_config = bitrate_config}; + RtpTransportConfig transport_config{ + .env = env, + .bitrate_config = bitrate_config, + .worker_thread = time_controller.GetMainThread()}; RtpTransportControllerSend transport_controller(transport_config); transport_controller.EnsureStarted();
diff --git a/media/engine/webrtc_video_engine_unittest.cc b/media/engine/webrtc_video_engine_unittest.cc index 08d0abc..c6d74b5 100644 --- a/media/engine/webrtc_video_engine_unittest.cc +++ b/media/engine/webrtc_video_engine_unittest.cc
@@ -387,7 +387,7 @@ env_(CreateEnvironment(field_trials_.CreateCopy(), time_controller_.CreateTaskQueueFactory(), time_controller_.GetClock())), - call_(Call::Create(CallConfig(env_))), + call_(Call::Create(CallConfig::CreateSingleThreaded(env_))), encoder_factory_(new FakeWebRtcVideoEncoderFactory), decoder_factory_(new FakeWebRtcVideoDecoderFactory), video_bitrate_allocator_factory_( @@ -1550,8 +1550,8 @@ // Create a call. GlobalSimulatedTimeController time_controller(Timestamp::Millis(4711)); const Environment env = CreateTestEnvironment({.time = &time_controller}); - CallConfig call_config(env); - const std::unique_ptr<Call> call = Call::Create(std::move(call_config)); + const std::unique_ptr<Call> call = + Call::Create(CallConfig::CreateSingleThreaded(env)); // Create send channel. const int send_ssrc = 123; @@ -1683,7 +1683,7 @@ : field_trials_(CreateTestFieldTrials()), env_(CreateTestEnvironment( {.field_trials = &field_trials_, .time = &time_controller_})), - call_(Call::Create(CallConfig(env_))), + call_(Call::Create(CallConfig::CreateSingleThreaded(env_))), video_bitrate_allocator_factory_( CreateBuiltinVideoBitrateAllocatorFactory()), engine_( @@ -1873,7 +1873,7 @@ void SetUp() override { // One testcase calls SetUp in a loop, only create call_ once. if (!call_) { - call_ = Call::Create(CallConfig(env_)); + call_ = Call::Create(CallConfig::CreateSingleThreaded(env_)); } MediaConfig media_config;
diff --git a/media/engine/webrtc_voice_engine_unittest.cc b/media/engine/webrtc_voice_engine_unittest.cc index 33b4a52..d41739cd 100644 --- a/media/engine/webrtc_voice_engine_unittest.cc +++ b/media/engine/webrtc_voice_engine_unittest.cc
@@ -3586,7 +3586,8 @@ env, adm, MockAudioEncoderFactory::CreateUnusedFactory(), MockAudioDecoderFactory::CreateUnusedFactory(), nullptr, apm, nullptr); AutoInitTerminate init_term(engine); - std::unique_ptr<Call> call = Call::Create(CallConfig(env)); + std::unique_ptr<Call> call = + Call::Create(CallConfig::CreateSingleThreaded(env)); std::unique_ptr<VoiceMediaSendChannelInterface> send_channel = engine.CreateSendChannel(env, call.get(), MediaConfig(), AudioOptions(), CryptoOptions()); @@ -3613,7 +3614,8 @@ MockAudioDecoderFactory::CreateUnusedFactory(), nullptr, apm, nullptr); AutoInitTerminate init_term(engine); - std::unique_ptr<Call> call = Call::Create(CallConfig(env)); + std::unique_ptr<Call> call = + Call::Create(CallConfig::CreateSingleThreaded(env)); std::unique_ptr<VoiceMediaSendChannelInterface> send_channel = engine.CreateSendChannel(env, call.get(), MediaConfig(), AudioOptions(), CryptoOptions()); @@ -3688,7 +3690,8 @@ env, adm, MockAudioEncoderFactory::CreateUnusedFactory(), MockAudioDecoderFactory::CreateUnusedFactory(), nullptr, apm, nullptr); AutoInitTerminate init_term(engine); - std::unique_ptr<Call> call = Call::Create(CallConfig(env)); + std::unique_ptr<Call> call = + Call::Create(CallConfig::CreateSingleThreaded(env)); std::vector<std::unique_ptr<VoiceMediaSendChannelInterface>> channels; while (channels.size() < 32) { @@ -3724,7 +3727,8 @@ env, adm, MockAudioEncoderFactory::CreateUnusedFactory(), CreateBuiltinAudioDecoderFactory(), nullptr, apm, nullptr); AutoInitTerminate init_term(engine); - std::unique_ptr<Call> call = Call::Create(CallConfig(env)); + std::unique_ptr<Call> call = + Call::Create(CallConfig::CreateSingleThreaded(env)); WebRtcVoiceReceiveChannel channel(env, &engine, MediaConfig(), AudioOptions(), CryptoOptions(), call.get()); @@ -3744,7 +3748,7 @@ CreateBuiltinAudioDecoderFactory(), nullptr, nullptr, nullptr); AutoInitTerminate init_term(engine); - CallConfig call_config(env); + CallConfig call_config = CallConfig::CreateSingleThreaded(env); { AudioState::Config config; config.audio_mixer = AudioMixerImpl::Create();
diff --git a/modules/pacing/task_queue_paced_sender.cc b/modules/pacing/task_queue_paced_sender.cc index 5646472..574388d 100644 --- a/modules/pacing/task_queue_paced_sender.cc +++ b/modules/pacing/task_queue_paced_sender.cc
@@ -43,7 +43,8 @@ PacingController::PacketSender* packet_sender, const FieldTrialsView& field_trials, TimeDelta max_hold_back_window, - int max_hold_back_window_in_packets) + int max_hold_back_window_in_packets, + TaskQueueBase* task_queue) : clock_(clock), max_hold_back_window_(max_hold_back_window), max_hold_back_window_in_packets_(max_hold_back_window_in_packets), @@ -53,7 +54,8 @@ is_shutdown_(false), packet_size_(/*alpha=*/0.95), include_overhead_(false), - task_queue_(TaskQueueBase::Current()) { + task_queue_(task_queue) { + RTC_DCHECK(task_queue_); RTC_DCHECK_GE(max_hold_back_window_, PacingController::kMinSleepTime); }
diff --git a/modules/pacing/task_queue_paced_sender.h b/modules/pacing/task_queue_paced_sender.h index d7ca29c..e5adcb1 100644 --- a/modules/pacing/task_queue_paced_sender.h +++ b/modules/pacing/task_queue_paced_sender.h
@@ -53,7 +53,8 @@ PacingController::PacketSender* packet_sender, const FieldTrialsView& field_trials, TimeDelta max_hold_back_window, - int max_hold_back_window_in_packets); + int max_hold_back_window_in_packets, + TaskQueueBase* task_queue); ~TaskQueuePacedSender() override;
diff --git a/modules/pacing/task_queue_paced_sender_unittest.cc b/modules/pacing/task_queue_paced_sender_unittest.cc index 9d96aa4..f2b24c4 100644 --- a/modules/pacing/task_queue_paced_sender_unittest.cc +++ b/modules/pacing/task_queue_paced_sender_unittest.cc
@@ -128,7 +128,8 @@ FieldTrials trials = CreateTestFieldTrials(); TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); // Insert a number of packets, covering one second. static constexpr size_t kPacketsToSend = 42; @@ -172,7 +173,8 @@ FieldTrials trials = CreateTestFieldTrials(); TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); // Insert a number of packets, covering one second. static constexpr size_t kPacketsToSend = 42; @@ -217,7 +219,8 @@ FieldTrials trials = CreateTestFieldTrials(); TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); // Insert a number of packets to be sent 200ms apart. const size_t kPacketsPerSecond = 5; @@ -275,7 +278,8 @@ FieldTrials trials = CreateTestFieldTrials(); TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); const DataRate kPacingDataRate = DataRate::KilobitsPerSec(125); @@ -302,9 +306,9 @@ GlobalSimulatedTimeController time_controller(Timestamp::Millis(1234)); NiceMock<MockPacketRouter> packet_router; FieldTrials trials = CreateTestFieldTrials(); - TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, - kCoalescingWindow, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender pacer( + time_controller.GetClock(), &packet_router, trials, kCoalescingWindow, + TaskQueuePacedSender::kNoPacketHoldback, time_controller.GetMainThread()); // Set rates so one packet adds one ms of buffer level. const DataSize kPacketSize = DataSize::Bytes(kDefaultPacketSize); @@ -341,9 +345,9 @@ GlobalSimulatedTimeController time_controller(Timestamp::Millis(1234)); MockPacketRouter packet_router; FieldTrials trials = CreateTestFieldTrials(); - TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, - kCoalescingWindow, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender pacer( + time_controller.GetClock(), &packet_router, trials, kCoalescingWindow, + TaskQueuePacedSender::kNoPacketHoldback, time_controller.GetMainThread()); // Set rates so one packet adds one ms of buffer level. const DataSize kPacketSize = DataSize::Bytes(kDefaultPacketSize); @@ -381,7 +385,8 @@ NiceMock<MockPacketRouter> packet_router; TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); // Set rates so one packet adds 4ms of buffer level. const DataSize kPacketSize = DataSize::Bytes(kDefaultPacketSize); @@ -455,7 +460,8 @@ MockPacketRouter packet_router; TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); // Set rates so one packet adds 4ms of buffer level. const DataSize kPacketSize = DataSize::Bytes(kDefaultPacketSize); @@ -518,7 +524,8 @@ NiceMock<MockPacketRouter> packet_router; FieldTrials trials = CreateTestFieldTrials(); TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, - kFixedCoalescingWindow, kPacketBasedHoldback); + kFixedCoalescingWindow, kPacketBasedHoldback, + time_controller.GetMainThread()); // Set rates so one packet adds one ms of buffer level. const DataSize kPacketSize = DataSize::Bytes(kDefaultPacketSize); @@ -570,7 +577,8 @@ MockPacketRouter packet_router; FieldTrials trials = CreateTestFieldTrials(); TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, - kFixedCoalescingWindow, kPacketBasedHoldback); + kFixedCoalescingWindow, kPacketBasedHoldback, + time_controller.GetMainThread()); // Set rates so one packet adds one ms of buffer level. const DataSize kPacketSize = DataSize::Bytes(kDefaultPacketSize); @@ -620,7 +628,8 @@ MockPacketRouter packet_router; TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); // Set rates so 2 packets adds 1ms of buffer level. const DataSize kPacketSize = DataSize::Bytes(kDefaultPacketSize); @@ -669,7 +678,8 @@ MockPacketRouter packet_router; TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); static constexpr DataRate kPacingRate = DataRate::BytesPerSec(kDefaultPacketSize * 10); @@ -712,7 +722,8 @@ FieldTrials trials = CreateTestFieldTrials(); TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); // Simulate ~2mbps video stream, covering one second. static constexpr size_t kPacketsToSend = 200; @@ -785,7 +796,8 @@ MockPacketRouter packet_router; TaskQueuePacedSender pacer(time_controller.GetClock(), &packet_router, trials, PacingController::kMinSleepTime, - TaskQueuePacedSender::kNoPacketHoldback); + TaskQueuePacedSender::kNoPacketHoldback, + time_controller.GetMainThread()); pacer.EnsureStarted(); pacer.SetConfig(PacerConfig::Create( time_controller.GetClock()->CurrentTime(),
diff --git a/pc/peer_connection_factory.cc b/pc/peer_connection_factory.cc index 288b889..ed12a4b 100644 --- a/pc/peer_connection_factory.cc +++ b/pc/peer_connection_factory.cc
@@ -362,7 +362,7 @@ const PeerConnectionInterface::RTCConfiguration& configuration) { RTC_DCHECK_RUN_ON(worker_thread()); - CallConfig call_config(env, network_thread()); + CallConfig call_config(env, worker_thread(), network_thread()); if (!context_->media_engine() || !context_->call_factory()) { return nullptr; }
diff --git a/rtc_tools/video_replay.cc b/rtc_tools/video_replay.cc index 14465cb..f4adfc9 100644 --- a/rtc_tools/video_replay.cc +++ b/rtc_tools/video_replay.cc
@@ -528,7 +528,8 @@ "worker_thread", TaskQueueFactory::Priority::kNormal); Event event; worker_thread_->PostTask([&]() { - call_ = Call::Create(CallConfig(env_)); + call_ = Call::Create(CallConfig::CreateWithJoinedWorkerAndNetworkQueue( + env_, worker_thread_.get())); // Creation of the streams must happen inside a task queue because it is // resued as a worker thread.
diff --git a/test/call_test.cc b/test/call_test.cc index 8594aa6..c47081d 100644 --- a/test/call_test.cc +++ b/test/call_test.cc
@@ -33,6 +33,7 @@ #include "api/rtp_headers.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_frame_generator.h" #include "api/test/simulated_network.h" @@ -259,20 +260,27 @@ }); } -CallConfig CallTest::SendCallConfig() const { - CallConfig sender_config(send_env_, network_thread_.get()); +CallConfig CallTest::SendCallConfig(TaskQueueBase* worker_task_queue) const { + if (worker_task_queue == nullptr) { + worker_task_queue = task_queue_.get(); + } + CallConfig sender_config(send_env_, worker_task_queue, network_thread_.get()); sender_config.network_state_predictor_factory = network_state_predictor_factory_.get(); sender_config.network_controller_factory = network_controller_factory_.get(); return sender_config; } -CallConfig CallTest::RecvCallConfig() const { - return CallConfig(recv_env_, network_thread_.get()); +CallConfig CallTest::RecvCallConfig(TaskQueueBase* worker_task_queue) const { + if (worker_task_queue == nullptr) { + worker_task_queue = task_queue_.get(); + } + return CallConfig(recv_env_, worker_task_queue, network_thread_.get()); } -void CallTest::CreateCalls() { - CreateCalls(SendCallConfig(), RecvCallConfig()); +void CallTest::CreateCalls(TaskQueueBase* worker_task_queue) { + CreateCalls(SendCallConfig(worker_task_queue), + RecvCallConfig(worker_task_queue)); } void CallTest::CreateCalls(CallConfig sender_config, @@ -281,23 +289,61 @@ CreateReceiverCall(std::move(receiver_config)); } -void CallTest::CreateSenderCall() { - CreateSenderCall(SendCallConfig()); +void CallTest::CreateSenderCall(TaskQueueBase* worker_task_queue) { + CreateSenderCall(SendCallConfig(worker_task_queue)); } void CallTest::CreateSenderCall(CallConfig config) { - sender_call_ = Call::Create(std::move(config)); + TaskQueueBase* worker = config.worker_task_queue; + if (worker->IsCurrent()) { + sender_call_ = Call::Create(std::move(config)); + } else { + SendTask(worker, [this, config = std::move(config)]() mutable { + sender_call_ = Call::Create(std::move(config)); + }); + } +} + +void CallTest::CreateReceiverCall(TaskQueueBase* worker_task_queue) { + CreateReceiverCall(RecvCallConfig(worker_task_queue)); } void CallTest::CreateReceiverCall(CallConfig config) { - receiver_call_ = Call::Create(std::move(config)); + TaskQueueBase* worker = config.worker_task_queue; + if (worker->IsCurrent()) { + receiver_call_ = Call::Create(std::move(config)); + } else { + SendTask(worker, [this, config = std::move(config)]() mutable { + receiver_call_ = Call::Create(std::move(config)); + }); + } } void CallTest::DestroyCalls() { - send_transport_.reset(); - receive_transport_.reset(); - sender_call_.reset(); - receiver_call_.reset(); + if (sender_call_) { + TaskQueueBase* worker = sender_call_->worker_thread(); + if (worker->IsCurrent()) { + send_transport_.reset(); + sender_call_.reset(); + } else { + SendTask(worker, [this]() { + send_transport_.reset(); + sender_call_.reset(); + }); + } + } + if (receiver_call_) { + TaskQueueBase* worker = receiver_call_->worker_thread(); + if (worker->IsCurrent()) { + receive_transport_.reset(); + receiver_call_.reset(); + } else { + SendTask(worker, [this]() { + receive_transport_.reset(); + receiver_call_.reset(); + }); + } + } } void CallTest::CreateVideoSendConfig(VideoSendStream::Config* video_config,
diff --git a/test/call_test.h b/test/call_test.h index e1ac4ed..b618dd9 100644 --- a/test/call_test.h +++ b/test/call_test.h
@@ -92,13 +92,14 @@ // to simplify test code. void RunBaseTest(BaseTest* test); - CallConfig SendCallConfig() const; - CallConfig RecvCallConfig() const; + CallConfig SendCallConfig(TaskQueueBase* worker_task_queue = nullptr) const; + CallConfig RecvCallConfig(TaskQueueBase* worker_task_queue = nullptr) const; - void CreateCalls(); + void CreateCalls(TaskQueueBase* worker_task_queue = nullptr); void CreateCalls(CallConfig sender_config, CallConfig receiver_config); - void CreateSenderCall(); + void CreateSenderCall(TaskQueueBase* worker_task_queue = nullptr); void CreateSenderCall(CallConfig config); + void CreateReceiverCall(TaskQueueBase* worker_task_queue = nullptr); void CreateReceiverCall(CallConfig config); void DestroyCalls(); Thread* network_thread() const { return network_thread_.get(); }
diff --git a/test/fuzzers/utils/BUILD.gn b/test/fuzzers/utils/BUILD.gn index 07671b0..be387f9 100644 --- a/test/fuzzers/utils/BUILD.gn +++ b/test/fuzzers/utils/BUILD.gn
@@ -26,6 +26,7 @@ "../../../api:time_controller", "../../../api:transport_api", "../../../api/environment", + "../../../api/task_queue", "../../../api/units:time_delta", "../../../api/units:timestamp", "../../../api/video:video_frame",
diff --git a/test/fuzzers/utils/rtp_replayer.cc b/test/fuzzers/utils/rtp_replayer.cc index 6d93f0d..c0a6ae1 100644 --- a/test/fuzzers/utils/rtp_replayer.cc +++ b/test/fuzzers/utils/rtp_replayer.cc
@@ -22,6 +22,7 @@ #include "api/call/transport.h" #include "api/environment/environment.h" #include "api/media_types.h" +#include "api/task_queue/task_queue_base.h" #include "api/test/time_controller.h" #include "api/units/time_delta.h" #include "api/units/timestamp.h" @@ -90,8 +91,8 @@ // by chromium. To avoid blocking when running in chromium, real (default) // task queues are used, while `time_controller` is used only for the Clock. Environment env = CreateTestEnvironment({.time = time_controller.GetClock()}); - CallConfig call_config(env); - std::unique_ptr<Call> call = Call::Create(std::move(call_config)); + std::unique_ptr<Call> call = + Call::Create(CallConfig::CreateSingleThreaded(env)); SetupVideoStreams(&receive_stream_configs, stream_state.get(), call.get()); // Start replaying the provided stream now that it has been configured.
diff --git a/test/scenario/call_client.cc b/test/scenario/call_client.cc index 4f30076..b3368e7 100644 --- a/test/scenario/call_client.cc +++ b/test/scenario/call_client.cc
@@ -91,7 +91,7 @@ CallClientConfig config, LoggingNetworkControllerFactory* network_controller_factory, scoped_refptr<AudioState> audio_state) { - CallConfig call_config(env); + CallConfig call_config = CallConfig::CreateSingleThreaded(env); call_config.bitrate_config.max_bitrate_bps = config.transport.rates.max_rate.bps_or(-1); call_config.bitrate_config.min_bitrate_bps =
diff --git a/video/BUILD.gn b/video/BUILD.gn index 1359128..5aada02 100644 --- a/video/BUILD.gn +++ b/video/BUILD.gn
@@ -974,6 +974,7 @@ "../api:frame_generator_api", "../api:rtp_parameters", "../api:simulated_network_api", + "../api/task_queue", "../api/video:video_frame", "../rtc_base:rtc_event", "../rtc_base:task_queue_for_test",
diff --git a/video/call_stats2.cc b/video/call_stats2.cc index 1d2a0f3..ad97cdd 100644 --- a/video/call_stats2.cc +++ b/video/call_stats2.cc
@@ -79,7 +79,6 @@ time_of_first_rtt_ms_(-1), task_queue_(task_queue) { RTC_DCHECK(task_queue_); - RTC_DCHECK_RUN_ON(task_queue_); } CallStats::~CallStats() {
diff --git a/video/end_to_end_tests/call_operation_tests.cc b/video/end_to_end_tests/call_operation_tests.cc index 272f661..340c01e 100644 --- a/video/end_to_end_tests/call_operation_tests.cc +++ b/video/end_to_end_tests/call_operation_tests.cc
@@ -12,6 +12,7 @@ #include <optional> #include "api/rtp_parameters.h" +#include "api/task_queue/task_queue_base.h" #include "api/test/create_frame_generator.h" #include "api/test/frame_generator_interface.h" #include "api/test/simulated_network.h" @@ -32,7 +33,7 @@ class CallOperationEndToEndTest : public test::CallTest {}; TEST_F(CallOperationEndToEndTest, ReceiverCanBeStartedTwice) { - CreateCalls(); + CreateCalls(TaskQueueBase::Current()); test::NullTransport transport; CreateSendConfig(1, 0, 0, &transport); @@ -47,7 +48,7 @@ } TEST_F(CallOperationEndToEndTest, ReceiverCanBeStoppedTwice) { - CreateCalls(); + CreateCalls(TaskQueueBase::Current()); test::NullTransport transport; CreateSendConfig(1, 0, 0, &transport); @@ -62,7 +63,7 @@ } TEST_F(CallOperationEndToEndTest, ReceiverCanBeStoppedAndRestarted) { - CreateCalls(); + CreateCalls(TaskQueueBase::Current()); test::NullTransport transport; CreateSendConfig(1, 0, 0, &transport);
diff --git a/video/end_to_end_tests/multi_stream_tester.cc b/video/end_to_end_tests/multi_stream_tester.cc index 49442f4..75302a2 100644 --- a/video/end_to_end_tests/multi_stream_tester.cc +++ b/video/end_to_end_tests/multi_stream_tester.cc
@@ -60,8 +60,11 @@ // to make test more stable. auto network_thread = Thread::CreateWithSocketServer(); network_thread->Start(); - CallConfig sender_config(env); - CallConfig receiver_config(env); + CallConfig sender_config = CallConfig::CreateWithJoinedWorkerAndNetworkQueue( + env, network_thread.get()); + CallConfig receiver_config = + CallConfig::CreateWithJoinedWorkerAndNetworkQueue(env, + network_thread.get()); std::unique_ptr<Call> sender_call; std::unique_ptr<Call> receiver_call; std::unique_ptr<test::DirectTransport> sender_transport;
diff --git a/video/video_send_stream_tests.cc b/video/video_send_stream_tests.cc index 73e8ecf..0562677 100644 --- a/video/video_send_stream_tests.cc +++ b/video/video_send_stream_tests.cc
@@ -2104,7 +2104,7 @@ int start_bitrate_kbps_ RTC_GUARDED_BY(mutex_); }; - CreateSenderCall(); + CreateSenderCall(TaskQueueBase::Current()); test::NullTransport transport; CreateSendConfig(1, 0, 0, &transport); @@ -3584,7 +3584,7 @@ void VideoSendStreamTest::TestRequestSourceRotateVideo( bool support_orientation_ext) { - CreateSenderCall(); + CreateSenderCall(TaskQueueBase::Current()); test::NullTransport transport; CreateSendConfig(1, 0, 0, &transport);