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 91 matching lines @@
       return nullptr;
   }
 
   void UpdateSendHistograms() EXCLUSIVE_LOCKS_REQUIRED(&bitrate_crit_);
   void UpdateReceiveHistograms();
   void UpdateAggregateNetworkState();
 
   Clock* const clock_;
 
   const int num_cpu_cores_;
+  // TODO(perkj): |worker_queu_| is supposed to replace |module_process_thread_|
+  // in the long run.
+  rtc::TaskQueue worker_queu_;

[tommi, 2016/06/17 07:59:01]

worker_queue_

[perkj_webrtc, 2016/06/27 14:34:34]

Done.

   const std::unique_ptr<ProcessThread> module_process_thread_;
   const std::unique_ptr<ProcessThread> pacer_thread_;
   const std::unique_ptr<CallStats> call_stats_;
   const std::unique_ptr<BitrateAllocator> bitrate_allocator_;
   Call::Config config_;
   rtc::ThreadChecker configuration_thread_checker_;
 
   NetworkState audio_network_state_;
   NetworkState video_network_state_;
 
@@ skipping 48 matching lines @@
 
 Call* Call::Create(const Call::Config& config) {
   return new internal::Call(config);
 }
 
 namespace internal {
 
 Call::Call(const Call::Config& config)
     : clock_(Clock::GetRealTimeClock()),
       num_cpu_cores_(CpuInfo::DetectNumberOfCores()),
+      worker_queu_("worker_queue"),
       module_process_thread_(ProcessThread::Create("ModuleProcessThread")),
       pacer_thread_(ProcessThread::Create("PacerThread")),
       call_stats_(new CallStats(clock_)),
       bitrate_allocator_(new BitrateAllocator(this)),
       config_(config),
       audio_network_state_(kNetworkUp),
       video_network_state_(kNetworkUp),
       receive_crit_(RWLockWrapper::CreateRWLock()),
       send_crit_(RWLockWrapper::CreateRWLock()),
       received_video_bytes_(0),
@@ skipping 198 matching lines @@
 webrtc::VideoSendStream* Call::CreateVideoSendStream(
     const webrtc::VideoSendStream::Config& config,
     const VideoEncoderConfig& encoder_config) {
   TRACE_EVENT0("webrtc", "Call::CreateVideoSendStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
   video_send_delay_stats_->AddSsrcs(config);
   // TODO(mflodman): Base the start bitrate on a current bandwidth estimate, if
   // the call has already started.
   VideoSendStream* send_stream = new VideoSendStream(
-      num_cpu_cores_, module_process_thread_.get(), call_stats_.get(),
-      congestion_controller_.get(), bitrate_allocator_.get(),
+      num_cpu_cores_, module_process_thread_.get(), &worker_queu_,
+      call_stats_.get(), congestion_controller_.get(), bitrate_allocator_.get(),
       video_send_delay_stats_.get(), &remb_, event_log_, config, encoder_config,
       suspended_video_send_ssrcs_);
   {
     WriteLockScoped write_lock(*send_crit_);
     for (uint32_t ssrc : config.rtp.ssrcs) {
       RTC_DCHECK(video_send_ssrcs_.find(ssrc) == video_send_ssrcs_.end());
       video_send_ssrcs_[ssrc] = send_stream;
     }
     video_send_streams_.insert(send_stream);
   }
@@ skipping 20 matching lines @@
         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 198 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_queu_|.
+  if (!worker_queu_.IsCurrent()) {
+    worker_queu_.PostTask([this, target_bitrate_bps, fraction_loss, rtt_ms] {
+      OnNetworkChanged(target_bitrate_bps, fraction_loss, rtt_ms);
+    });
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&worker_queu_);
   bitrate_allocator_->OnNetworkChanged(target_bitrate_bps, fraction_loss,
                                        rtt_ms);
 
   {
     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;
     // Pacer bitrate might be higher than bitrate estimate if enforcing min
     // bitrate.
@@ skipping 156 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