Add task queue to Call.

webrtc/call/call.cc

 /*
  *  Copyright (c) 2013 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
 #include <string.h>
-
 #include <algorithm>
 #include <map>
 #include <memory>
 #include <vector>
 
 #include "webrtc/audio/audio_receive_stream.h"
 #include "webrtc/audio/audio_send_stream.h"
 #include "webrtc/audio/audio_state.h"
 #include "webrtc/audio/scoped_voe_interface.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/constructormagic.h"
 #include "webrtc/base/logging.h"
+#include "webrtc/base/task_queue.h"
 #include "webrtc/base/thread_annotations.h"
 #include "webrtc/base/thread_checker.h"
 #include "webrtc/base/trace_event.h"
 #include "webrtc/call.h"
 #include "webrtc/call/bitrate_allocator.h"
 #include "webrtc/call/rtc_event_log.h"
 #include "webrtc/config.h"
 #include "webrtc/modules/bitrate_controller/include/bitrate_controller.h"
 #include "webrtc/modules/congestion_controller/include/congestion_controller.h"
 #include "webrtc/modules/pacing/paced_sender.h"
@@ skipping 32 matching lines @@
   webrtc::AudioSendStream* CreateAudioSendStream(
       const webrtc::AudioSendStream::Config& config) override;
   void DestroyAudioSendStream(webrtc::AudioSendStream* send_stream) override;
 
   webrtc::AudioReceiveStream* CreateAudioReceiveStream(
       const webrtc::AudioReceiveStream::Config& config) override;
   void DestroyAudioReceiveStream(
       webrtc::AudioReceiveStream* receive_stream) override;
 
   webrtc::VideoSendStream* CreateVideoSendStream(
-      const webrtc::VideoSendStream::Config& config,
-      const VideoEncoderConfig& encoder_config) override;
+      webrtc::VideoSendStream::Config config,
+      VideoEncoderConfig encoder_config) override;
   void DestroyVideoSendStream(webrtc::VideoSendStream* send_stream) override;
 
   webrtc::VideoReceiveStream* CreateVideoReceiveStream(
       webrtc::VideoReceiveStream::Config configuration) override;
   void DestroyVideoReceiveStream(
       webrtc::VideoReceiveStream* receive_stream) override;
 
   Stats GetStats() const override;
 
   DeliveryStatus DeliverPacket(MediaType media_type,
@@ skipping 102 matching lines @@
   uint32_t min_allocated_send_bitrate_bps_ GUARDED_BY(&bitrate_crit_);
   int64_t num_bitrate_updates_ GUARDED_BY(&bitrate_crit_);
   uint32_t configured_max_padding_bitrate_bps_ GUARDED_BY(&bitrate_crit_);
 
   std::map<std::string, rtc::NetworkRoute> network_routes_;
 
   VieRemb remb_;
   const std::unique_ptr<CongestionController> congestion_controller_;
   const std::unique_ptr<SendDelayStats> video_send_delay_stats_;
   const int64_t start_ms_;
+  // TODO(perkj): |worker_queue_| is supposed to replace
+  // |module_process_thread_|.
+  // |worker_queue| is defined last to ensure all pending tasks are cancelled
+  // and deleted before any other members.
+  rtc::TaskQueue worker_queue_;
 
   RTC_DISALLOW_COPY_AND_ASSIGN(Call);
 };
 }  // namespace internal
 
 std::string Call::Stats::ToString(int64_t time_ms) const {
   std::stringstream ss;
   ss << "Call stats: " << time_ms << ", {";
   ss << "send_bw_bps: " << send_bandwidth_bps << ", ";
   ss << "recv_bw_bps: " << recv_bandwidth_bps << ", ";
@@ skipping 31 matching lines @@
       first_packet_sent_ms_(-1),
       estimated_send_bitrate_sum_kbits_(0),
       pacer_bitrate_sum_kbits_(0),
       min_allocated_send_bitrate_bps_(0),
       num_bitrate_updates_(0),
       configured_max_padding_bitrate_bps_(0),
       remb_(clock_),
       congestion_controller_(
           new CongestionController(clock_, this, &remb_, event_log_.get())),
       video_send_delay_stats_(new SendDelayStats(clock_)),
-      start_ms_(clock_->TimeInMilliseconds()) {
+      start_ms_(clock_->TimeInMilliseconds()),
+      worker_queue_("call_worker_queue") {
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   RTC_DCHECK_GE(config.bitrate_config.min_bitrate_bps, 0);
   RTC_DCHECK_GE(config.bitrate_config.start_bitrate_bps,
                 config.bitrate_config.min_bitrate_bps);
   if (config.bitrate_config.max_bitrate_bps != -1) {
     RTC_DCHECK_GE(config.bitrate_config.max_bitrate_bps,
                   config.bitrate_config.start_bitrate_bps);
   }
 
   Trace::CreateTrace();
   call_stats_->RegisterStatsObserver(congestion_controller_.get());
 
   congestion_controller_->SetBweBitrates(
       config_.bitrate_config.min_bitrate_bps,
       config_.bitrate_config.start_bitrate_bps,
       config_.bitrate_config.max_bitrate_bps);
 
   module_process_thread_->Start();
   module_process_thread_->RegisterModule(call_stats_.get());
   module_process_thread_->RegisterModule(congestion_controller_.get());
   pacer_thread_->RegisterModule(congestion_controller_->pacer());
   pacer_thread_->RegisterModule(
       congestion_controller_->GetRemoteBitrateEstimator(true));
   pacer_thread_->Start();
 }
 
 Call::~Call() {
   RTC_DCHECK(!remb_.InUse());
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
+
   RTC_CHECK(audio_send_ssrcs_.empty());
   RTC_CHECK(video_send_ssrcs_.empty());
   RTC_CHECK(video_send_streams_.empty());
   RTC_CHECK(audio_receive_ssrcs_.empty());
   RTC_CHECK(video_receive_ssrcs_.empty());
   RTC_CHECK(video_receive_streams_.empty());
 
   pacer_thread_->Stop();
   pacer_thread_->DeRegisterModule(congestion_controller_->pacer());
   pacer_thread_->DeRegisterModule(
       congestion_controller_->GetRemoteBitrateEstimator(true));
   module_process_thread_->DeRegisterModule(congestion_controller_.get());
   module_process_thread_->DeRegisterModule(call_stats_.get());
   module_process_thread_->Stop();
   call_stats_->DeregisterStatsObserver(congestion_controller_.get());
 
   // Only update histograms after process threads have been shut down, so that
   // they won't try to concurrently update stats.
-  UpdateSendHistograms();
+  {
+    rtc::CritScope lock(&bitrate_crit_);
+    UpdateSendHistograms();
+  }
   UpdateReceiveHistograms();
   UpdateHistograms();
 
   Trace::ReturnTrace();
 }
 
 void Call::UpdateHistograms() {
   RTC_LOGGED_HISTOGRAM_COUNTS_100000(
       "WebRTC.Call.LifetimeInSeconds",
       (clock_->TimeInMilliseconds() - start_ms_) / 1000);
@@ skipping 51 matching lines @@
   // thread. Re-enable once that is fixed.
   // RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   return this;
 }
 
 webrtc::AudioSendStream* Call::CreateAudioSendStream(
     const webrtc::AudioSendStream::Config& config) {
   TRACE_EVENT0("webrtc", "Call::CreateAudioSendStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   AudioSendStream* send_stream = new AudioSendStream(
-      config, config_.audio_state, congestion_controller_.get(),
+      config, config_.audio_state, &worker_queue_, congestion_controller_.get(),
       bitrate_allocator_.get());
   {
     WriteLockScoped write_lock(*send_crit_);
     RTC_DCHECK(audio_send_ssrcs_.find(config.rtp.ssrc) ==
                audio_send_ssrcs_.end());
     audio_send_ssrcs_[config.rtp.ssrc] = send_stream;
   }
   send_stream->SignalNetworkState(audio_network_state_);
   UpdateAggregateNetworkState();
   return send_stream;
@@ skipping 55 matching lines @@
         it->second == audio_receive_stream) {
       sync_stream_mapping_.erase(it);
       ConfigureSync(sync_group);
     }
   }
   UpdateAggregateNetworkState();
   delete audio_receive_stream;
 }
 
 webrtc::VideoSendStream* Call::CreateVideoSendStream(
-    const webrtc::VideoSendStream::Config& config,
-    const VideoEncoderConfig& encoder_config) {
+    webrtc::VideoSendStream::Config config,
+    VideoEncoderConfig encoder_config) {
   TRACE_EVENT0("webrtc", "Call::CreateVideoSendStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
   video_send_delay_stats_->AddSsrcs(config);
+  event_log_->LogVideoSendStreamConfig(config);
+
   // TODO(mflodman): Base the start bitrate on a current bandwidth estimate, if
   // the call has already started.
+  // Copy ssrcs from |config| since |config| is moved.
+  std::vector<uint32_t> ssrcs = config.rtp.ssrcs;
   VideoSendStream* send_stream = new VideoSendStream(
-      num_cpu_cores_, module_process_thread_.get(), call_stats_.get(),
-      congestion_controller_.get(), bitrate_allocator_.get(),
-      video_send_delay_stats_.get(), &remb_, event_log_.get(), config,
-      encoder_config, suspended_video_send_ssrcs_);
+      num_cpu_cores_, module_process_thread_.get(), &worker_queue_,
+      call_stats_.get(), congestion_controller_.get(), bitrate_allocator_.get(),
+      video_send_delay_stats_.get(), &remb_, event_log_.get(),
+      std::move(config), std::move(encoder_config),
+      suspended_video_send_ssrcs_);
+
   {
     WriteLockScoped write_lock(*send_crit_);
-    for (uint32_t ssrc : config.rtp.ssrcs) {
+    for (uint32_t ssrc : ssrcs) {
       RTC_DCHECK(video_send_ssrcs_.find(ssrc) == video_send_ssrcs_.end());
       video_send_ssrcs_[ssrc] = send_stream;
     }
     video_send_streams_.insert(send_stream);
   }
   send_stream->SignalNetworkState(video_network_state_);
   UpdateAggregateNetworkState();
-  event_log_->LogVideoSendStreamConfig(config);
+
   return send_stream;
 }
 
 void Call::DestroyVideoSendStream(webrtc::VideoSendStream* send_stream) {
   TRACE_EVENT0("webrtc", "Call::DestroyVideoSendStream");
   RTC_DCHECK(send_stream != nullptr);
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
   send_stream->Stop();
 
   VideoSendStream* send_stream_impl = nullptr;
   {
     WriteLockScoped write_lock(*send_crit_);
     auto it = video_send_ssrcs_.begin();
     while (it != video_send_ssrcs_.end()) {
       if (it->second == static_cast<VideoSendStream*>(send_stream)) {
         send_stream_impl = it->second;
         video_send_ssrcs_.erase(it++);
       } else {
         ++it;
       }
     }
     video_send_streams_.erase(send_stream_impl);
   }
   RTC_CHECK(send_stream_impl != nullptr);
 
-  VideoSendStream::RtpStateMap rtp_state = send_stream_impl->GetRtpStates();
+  VideoSendStream::RtpStateMap rtp_state =
+      send_stream_impl->StopPermanentlyAndGetRtpStates();
 
   for (VideoSendStream::RtpStateMap::iterator it = rtp_state.begin();
-       it != rtp_state.end();
-       ++it) {
+       it != rtp_state.end(); ++it) {
     suspended_video_send_ssrcs_[it->first] = it->second;
   }
 
   UpdateAggregateNetworkState();
   delete send_stream_impl;
 }
 
 webrtc::VideoReceiveStream* Call::CreateVideoReceiveStream(
     webrtc::VideoReceiveStream::Config configuration) {
   TRACE_EVENT0("webrtc", "Call::CreateVideoReceiveStream");
@@ skipping 209 matching lines @@
 void Call::OnSentPacket(const rtc::SentPacket& sent_packet) {
   if (first_packet_sent_ms_ == -1)
     first_packet_sent_ms_ = clock_->TimeInMilliseconds();
   video_send_delay_stats_->OnSentPacket(sent_packet.packet_id,
                                         clock_->TimeInMilliseconds());
   congestion_controller_->OnSentPacket(sent_packet);
 }
 
 void Call::OnNetworkChanged(uint32_t target_bitrate_bps, uint8_t fraction_loss,
                             int64_t rtt_ms) {
+  // TODO(perkj): Consider making sure CongestionController operates on
+  // |worker_queue_|.
+  if (!worker_queue_.IsCurrent()) {
+    worker_queue_.PostTask([this, target_bitrate_bps, fraction_loss, rtt_ms] {
+      OnNetworkChanged(target_bitrate_bps, fraction_loss, rtt_ms);
+    });
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&worker_queue_);
   bitrate_allocator_->OnNetworkChanged(target_bitrate_bps, fraction_loss,
                                        rtt_ms);
 
   // Ignore updates where the bitrate is zero because the aggregate network
   // state is down.
   if (target_bitrate_bps > 0) {
     rtc::CritScope lock(&bitrate_crit_);
     // We only update these stats if we have send streams, and assume that
     // OnNetworkChanged is called roughly with a fixed frequency.
     estimated_send_bitrate_sum_kbits_ += target_bitrate_bps / 1000;
@@ skipping 159 matching lines @@
   // thread. Then this check can be enabled.
   // RTC_DCHECK(!configuration_thread_checker_.CalledOnValidThread());
   if (RtpHeaderParser::IsRtcp(packet, length))
     return DeliverRtcp(media_type, packet, length);
 
   return DeliverRtp(media_type, packet, length, packet_time);
 }
 
 }  // namespace internal
 }  // namespace webrtc