From 537012405b7e9452c0ce282d621023dd8a714a10 Mon Sep 17 00:00:00 2001 From: Sebastian Jansson Date: Mon, 19 Mar 2018 14:10:34 +0100 Subject: [PATCH] Replacing unique pointer with raw pointer in SSCC checks. Replacing the unique pointer used for access checks with a raw pointer pointing to the object owned by the unique pointer. This is to stop tsan from detecting a race between .get() done on the task queue and .reset() done in the destructor. Bug: webrtc:8415 Change-Id: Iae2ea9a2d38f319e73146e6b1e360b11b1708c76 Reviewed-on: https://webrtc-review.googlesource.com/62560 Reviewed-by: Stefan Holmer Reviewed-by: Niels Moller Commit-Queue: Sebastian Jansson Cr-Commit-Position: refs/heads/master@{#22492} --- .../include/send_side_congestion_controller.h | 37 +++++++++-------- .../rtp/send_side_congestion_controller.cc | 41 ++++++++++--------- 2 files changed, 43 insertions(+), 35 deletions(-) diff --git a/modules/congestion_controller/rtp/include/send_side_congestion_controller.h b/modules/congestion_controller/rtp/include/send_side_congestion_controller.h index 68e4843b54..9ce0826b2b 100644 --- a/modules/congestion_controller/rtp/include/send_side_congestion_controller.h +++ b/modules/congestion_controller/rtp/include/send_side_congestion_controller.h @@ -138,16 +138,16 @@ class SendSideCongestionController void WaitOnTasksForTest(); private: - void MaybeCreateControllers(); - void StartProcessPeriodicTasks() RTC_RUN_ON(task_queue_); - void UpdateControllerWithTimeInterval() RTC_RUN_ON(task_queue_); - void UpdatePacerQueue() RTC_RUN_ON(task_queue_); + void MaybeCreateControllers() RTC_RUN_ON(task_queue_ptr_); + void StartProcessPeriodicTasks() RTC_RUN_ON(task_queue_ptr_); + void UpdateControllerWithTimeInterval() RTC_RUN_ON(task_queue_ptr_); + void UpdatePacerQueue() RTC_RUN_ON(task_queue_ptr_); - void UpdateStreamsConfig() RTC_RUN_ON(task_queue_); + void UpdateStreamsConfig() RTC_RUN_ON(task_queue_ptr_); void MaybeUpdateOutstandingData(); void OnReceivedRtcpReceiverReportBlocks(const ReportBlockList& report_blocks, int64_t now_ms) - RTC_RUN_ON(task_queue_); + RTC_RUN_ON(task_queue_ptr_); const Clock* const clock_; // PacedSender is thread safe and doesn't need protection here. @@ -158,35 +158,40 @@ class SendSideCongestionController const std::unique_ptr controller_factory_; const std::unique_ptr pacer_controller_ - RTC_GUARDED_BY(task_queue_); + RTC_GUARDED_BY(task_queue_ptr_); std::unique_ptr control_handler_ - RTC_GUARDED_BY(task_queue_); + RTC_GUARDED_BY(task_queue_ptr_); std::unique_ptr controller_ - RTC_GUARDED_BY(task_queue_); + RTC_GUARDED_BY(task_queue_ptr_); - TimeDelta process_interval_ RTC_GUARDED_BY(task_queue_); + TimeDelta process_interval_ RTC_GUARDED_BY(task_queue_ptr_); std::map last_report_blocks_ - RTC_GUARDED_BY(task_queue_); - Timestamp last_report_block_time_ RTC_GUARDED_BY(task_queue_); + RTC_GUARDED_BY(task_queue_ptr_); + Timestamp last_report_block_time_ RTC_GUARDED_BY(task_queue_ptr_); - NetworkChangedObserver* observer_ RTC_GUARDED_BY(task_queue_); - NetworkControllerConfig initial_config_ RTC_GUARDED_BY(task_queue_); - StreamsConfig streams_config_ RTC_GUARDED_BY(task_queue_); + NetworkChangedObserver* observer_ RTC_GUARDED_BY(task_queue_ptr_); + NetworkControllerConfig initial_config_ RTC_GUARDED_BY(task_queue_ptr_); + StreamsConfig streams_config_ RTC_GUARDED_BY(task_queue_ptr_); const bool send_side_bwe_with_overhead_; // Transport overhead is written by OnNetworkRouteChanged and read by // AddPacket. // TODO(srte): Remove atomic when feedback adapter runs on task queue. std::atomic transport_overhead_bytes_per_packet_; - bool network_available_ RTC_GUARDED_BY(task_queue_); + bool network_available_ RTC_GUARDED_BY(task_queue_ptr_); // Protects access to last_packet_feedback_vector_ in feedback adapter. // TODO(srte): Remove this checker when feedback adapter runs on task queue. rtc::RaceChecker worker_race_; + // Caches task_queue_.get(), to avoid racing with destructor. + // Note that this is declared before task_queue_ to ensure that it is not + // invalidated until no more tasks can be running on the task queue. + rtc::TaskQueue* task_queue_ptr_; + // Note that moving ownership of the task queue makes it neccessary to make // sure that there is no outstanding tasks on it using destructed objects. // This is currently guranteed by using explicit reset in the destructor of diff --git a/modules/congestion_controller/rtp/send_side_congestion_controller.cc b/modules/congestion_controller/rtp/send_side_congestion_controller.cc index 7145fe3434..c6510b17c5 100644 --- a/modules/congestion_controller/rtp/send_side_congestion_controller.cc +++ b/modules/congestion_controller/rtp/send_side_congestion_controller.cc @@ -315,6 +315,7 @@ SendSideCongestionController::SendSideCongestionController( transport_overhead_bytes_per_packet_(0), network_available_(false), task_queue_(MakeUnique("SendSideCCQueue")) { + task_queue_ptr_ = task_queue_.get(); initial_config_.constraints = ConvertConstraints(min_bitrate_bps, max_bitrate_bps, clock_); initial_config_.stream_based_config = StreamsConfig(); @@ -333,7 +334,6 @@ SendSideCongestionController::SendSideCongestionController( // bandwidth is set before this class is initialized so the controllers can be // created in the constructor. void SendSideCongestionController::MaybeCreateControllers() { - RTC_DCHECK_RUN_ON(task_queue_.get()); if (controller_ || !network_available_ || !observer_) return; @@ -353,6 +353,9 @@ void SendSideCongestionController::MaybeCreateControllers() { SendSideCongestionController::~SendSideCongestionController() { // Must be destructed before any objects used by calls on the task queue. task_queue_.reset(); + // Singe the task queue has been destructed, it is now safe to reset + // task_queue_raw_ which is only used by tasks on the task queue. + task_queue_ptr_ = nullptr; } void SendSideCongestionController::RegisterPacketFeedbackObserver( @@ -368,7 +371,7 @@ void SendSideCongestionController::DeRegisterPacketFeedbackObserver( void SendSideCongestionController::RegisterNetworkObserver( NetworkChangedObserver* observer) { task_queue_->PostTask([this, observer]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); RTC_DCHECK(observer_ == nullptr); observer_ = observer; MaybeCreateControllers(); @@ -381,7 +384,7 @@ void SendSideCongestionController::SetBweBitrates(int min_bitrate_bps, TargetRateConstraints constraints = ConvertConstraints(min_bitrate_bps, max_bitrate_bps, clock_); task_queue_->PostTask([this, constraints, start_bitrate_bps]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); if (controller_) { controller_->OnTargetRateConstraints(constraints); } else { @@ -398,7 +401,7 @@ void SendSideCongestionController::SetAllocatedSendBitrateLimits( int64_t max_total_bitrate_bps) { task_queue_->PostTask([this, min_send_bitrate_bps, max_padding_bitrate_bps, max_total_bitrate_bps]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); streams_config_.min_pacing_rate = DataRate::bps(min_send_bitrate_bps); streams_config_.max_padding_rate = DataRate::bps(max_padding_bitrate_bps); streams_config_.max_total_allocated_bitrate = @@ -425,7 +428,7 @@ void SendSideCongestionController::OnNetworkRouteChanged( msg.starting_rate = start_bitrate_bps > 0 ? DataRate::bps(start_bitrate_bps) : DataRate::kNotInitialized; task_queue_->PostTask([this, msg]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); controller_->OnNetworkRouteChange(msg); pacer_controller_->OnNetworkRouteChange(msg); }); @@ -438,7 +441,7 @@ bool SendSideCongestionController::AvailableBandwidth( // running on the task queue. // TODO(srte): Remove this function when RtpTransportControllerSend stops // calling it. - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); if (!control_handler_) { return false; } @@ -458,7 +461,7 @@ RtcpBandwidthObserver* SendSideCongestionController::GetBandwidthObserver() { void SendSideCongestionController::EnablePeriodicAlrProbing(bool enable) { task_queue_->PostTask([this, enable]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); streams_config_.requests_alr_probing = enable; UpdateStreamsConfig(); }); @@ -482,7 +485,7 @@ void SendSideCongestionController::SignalNetworkState(NetworkState state) { msg.at_time = Timestamp::ms(clock_->TimeInMilliseconds()); msg.network_available = state == kNetworkUp; task_queue_->PostTask([this, msg]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); network_available_ = msg.network_available; if (controller_) { controller_->OnNetworkAvailability(msg); @@ -509,7 +512,7 @@ void SendSideCongestionController::OnSentPacket( msg.size = DataSize::bytes(packet->payload_size); msg.send_time = Timestamp::ms(packet->send_time_ms); task_queue_->PostTask([this, msg]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); if (controller_) controller_->OnSentPacket(msg); }); @@ -524,7 +527,7 @@ void SendSideCongestionController::OnRttUpdate(int64_t avg_rtt_ms, report.round_trip_time = TimeDelta::ms(avg_rtt_ms); report.smoothed = true; task_queue_->PostTask([this, report]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); if (controller_) controller_->OnRoundTripTimeUpdate(report); }); @@ -540,7 +543,7 @@ void SendSideCongestionController::Process() { } void SendSideCongestionController::StartProcessPeriodicTasks() { - task_queue_->PostDelayedTask( + task_queue_ptr_->PostDelayedTask( NewPeriodicTask( rtc::Bind( &SendSideCongestionController::UpdateControllerWithTimeInterval, @@ -548,7 +551,7 @@ void SendSideCongestionController::StartProcessPeriodicTasks() { process_interval_.ms()), process_interval_.ms()); - task_queue_->PostDelayedTask( + task_queue_ptr_->PostDelayedTask( NewPeriodicTask( rtc::Bind(&SendSideCongestionController::UpdatePacerQueue, this), PacerQueueUpdateIntervalMs), @@ -605,7 +608,7 @@ void SendSideCongestionController::OnTransportFeedback( msg.data_in_flight = DataSize::bytes(transport_feedback_adapter_.GetOutstandingBytes()); task_queue_->PostTask([this, msg]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); if (controller_) controller_->OnTransportPacketsFeedback(msg); }); @@ -617,7 +620,7 @@ void SendSideCongestionController::MaybeUpdateOutstandingData() { msg.in_flight_data = DataSize::bytes(transport_feedback_adapter_.GetOutstandingBytes()); task_queue_->PostTask([this, msg]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); pacer_controller_->OnOutstandingData(msg); }); } @@ -630,7 +633,7 @@ SendSideCongestionController::GetTransportFeedbackVector() const { void SendSideCongestionController::PostPeriodicTasksForTest() { task_queue_->PostTask([this]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); UpdateControllerWithTimeInterval(); UpdatePacerQueue(); }); @@ -644,7 +647,7 @@ void SendSideCongestionController::WaitOnTasksForTest() { void SendSideCongestionController::SetPacingFactor(float pacing_factor) { task_queue_->PostTask([this, pacing_factor]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); streams_config_.pacing_factor = pacing_factor; UpdateStreamsConfig(); }); @@ -656,7 +659,7 @@ void SendSideCongestionController::OnReceivedEstimatedBitrate( msg.receive_time = Timestamp::ms(clock_->TimeInMilliseconds()); msg.bandwidth = DataRate::bps(bitrate); task_queue_->PostTask([this, msg]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); if (controller_) controller_->OnRemoteBitrateReport(msg); }); @@ -667,12 +670,12 @@ void SendSideCongestionController::OnReceivedRtcpReceiverReport( int64_t rtt_ms, int64_t now_ms) { task_queue_->PostTask([this, report_blocks, now_ms]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); OnReceivedRtcpReceiverReportBlocks(report_blocks, now_ms); }); task_queue_->PostTask([this, now_ms, rtt_ms]() { - RTC_DCHECK_RUN_ON(task_queue_.get()); + RTC_DCHECK_RUN_ON(task_queue_ptr_); RoundTripTimeUpdate report; report.receive_time = Timestamp::ms(now_ms); report.round_trip_time = TimeDelta::ms(rtt_ms);