Add task queue to Call.

webrtc/video/vie_encoder.cc

 /*
  *  Copyright (c) 2012 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 "webrtc/video/vie_encoder.h"
 
 #include <algorithm>
 #include <limits>
 
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
 #include "webrtc/base/trace_event.h"
 #include "webrtc/base/timeutils.h"
 #include "webrtc/modules/pacing/paced_sender.h"
 #include "webrtc/modules/video_coding/include/video_coding.h"
 #include "webrtc/modules/video_coding/include/video_coding_defines.h"
 #include "webrtc/system_wrappers/include/metrics.h"
 #include "webrtc/video/overuse_frame_detector.h"
 #include "webrtc/video/send_statistics_proxy.h"
 #include "webrtc/video_frame.h"
 
 namespace webrtc {
 
+namespace {
+
+VideoCodecType PayloadNameToCodecType(const std::string& payload_name) {
+  if (payload_name == "VP8")
+    return kVideoCodecVP8;
+  if (payload_name == "VP9")
+    return kVideoCodecVP9;
+  if (payload_name == "H264")
+    return kVideoCodecH264;
+  return kVideoCodecGeneric;
+}
+
+VideoCodec VideoEncoderConfigToVideoCodec(const VideoEncoderConfig& config,
+                                          const std::string& payload_name,
+                                          int payload_type) {
+  const std::vector<VideoStream>& streams = config.streams;
+  static const int kEncoderMinBitrateKbps = 30;
+  RTC_DCHECK(!streams.empty());
+  RTC_DCHECK_GE(config.min_transmit_bitrate_bps, 0);
+
+  VideoCodec video_codec;
+  memset(&video_codec, 0, sizeof(video_codec));
+  video_codec.codecType = PayloadNameToCodecType(payload_name);
+
+  switch (config.content_type) {
+    case VideoEncoderConfig::ContentType::kRealtimeVideo:
+      video_codec.mode = kRealtimeVideo;
+      break;
+    case VideoEncoderConfig::ContentType::kScreen:
+      video_codec.mode = kScreensharing;
+      if (config.streams.size() == 1 &&
+          config.streams[0].temporal_layer_thresholds_bps.size() == 1) {
+        video_codec.targetBitrate =
+            config.streams[0].temporal_layer_thresholds_bps[0] / 1000;
+      }
+      break;
+  }
+
+  switch (video_codec.codecType) {
+    case kVideoCodecVP8: {
+      if (config.encoder_specific_settings) {
+        video_codec.codecSpecific.VP8 = *reinterpret_cast<const VideoCodecVP8*>(
+            config.encoder_specific_settings);
+      } else {
+        video_codec.codecSpecific.VP8 = VideoEncoder::GetDefaultVp8Settings();
+      }
+      video_codec.codecSpecific.VP8.numberOfTemporalLayers =
+          static_cast<unsigned char>(
+              streams.back().temporal_layer_thresholds_bps.size() + 1);
+      break;
+    }
+    case kVideoCodecVP9: {
+      if (config.encoder_specific_settings) {
+        video_codec.codecSpecific.VP9 = *reinterpret_cast<const VideoCodecVP9*>(
+            config.encoder_specific_settings);
+        if (video_codec.mode == kScreensharing) {
+          video_codec.codecSpecific.VP9.flexibleMode = true;
+          // For now VP9 screensharing use 1 temporal and 2 spatial layers.
+          RTC_DCHECK_EQ(video_codec.codecSpecific.VP9.numberOfTemporalLayers,
+                        1);
+          RTC_DCHECK_EQ(video_codec.codecSpecific.VP9.numberOfSpatialLayers, 2);
+        }
+      } else {
+        video_codec.codecSpecific.VP9 = VideoEncoder::GetDefaultVp9Settings();
+      }
+      video_codec.codecSpecific.VP9.numberOfTemporalLayers =
+          static_cast<unsigned char>(
+              streams.back().temporal_layer_thresholds_bps.size() + 1);
+      break;
+    }
+    case kVideoCodecH264: {
+      if (config.encoder_specific_settings) {
+        video_codec.codecSpecific.H264 =
+            *reinterpret_cast<const VideoCodecH264*>(
+                config.encoder_specific_settings);
+      } else {
+        video_codec.codecSpecific.H264 = VideoEncoder::GetDefaultH264Settings();
+      }
+      break;
+    }
+    default:
+      // TODO(pbos): Support encoder_settings codec-agnostically.
+      RTC_DCHECK(!config.encoder_specific_settings)
+          << "Encoder-specific settings for codec type not wired up.";
+      break;
+  }
+
+  strncpy(video_codec.plName, payload_name.c_str(), kPayloadNameSize - 1);
+  video_codec.plName[kPayloadNameSize - 1] = '\0';
+  video_codec.plType = payload_type;
+  video_codec.numberOfSimulcastStreams =
+      static_cast<unsigned char>(streams.size());
+  video_codec.minBitrate = streams[0].min_bitrate_bps / 1000;
+  if (video_codec.minBitrate < kEncoderMinBitrateKbps)
+    video_codec.minBitrate = kEncoderMinBitrateKbps;
+  RTC_DCHECK_LE(streams.size(), static_cast<size_t>(kMaxSimulcastStreams));
+  if (video_codec.codecType == kVideoCodecVP9) {
+    // If the vector is empty, bitrates will be configured automatically.
+    RTC_DCHECK(config.spatial_layers.empty() ||
+               config.spatial_layers.size() ==
+                   video_codec.codecSpecific.VP9.numberOfSpatialLayers);
+    RTC_DCHECK_LE(video_codec.codecSpecific.VP9.numberOfSpatialLayers,
+                  kMaxSimulcastStreams);
+    for (size_t i = 0; i < config.spatial_layers.size(); ++i)
+      video_codec.spatialLayers[i] = config.spatial_layers[i];
+  }
+  for (size_t i = 0; i < streams.size(); ++i) {
+    SimulcastStream* sim_stream = &video_codec.simulcastStream[i];
+    RTC_DCHECK_GT(streams[i].width, 0u);
+    RTC_DCHECK_GT(streams[i].height, 0u);
+    RTC_DCHECK_GT(streams[i].max_framerate, 0);
+    // Different framerates not supported per stream at the moment.
+    RTC_DCHECK_EQ(streams[i].max_framerate, streams[0].max_framerate);
+    RTC_DCHECK_GE(streams[i].min_bitrate_bps, 0);
+    RTC_DCHECK_GE(streams[i].target_bitrate_bps, streams[i].min_bitrate_bps);
+    RTC_DCHECK_GE(streams[i].max_bitrate_bps, streams[i].target_bitrate_bps);
+    RTC_DCHECK_GE(streams[i].max_qp, 0);
+
+    sim_stream->width = static_cast<uint16_t>(streams[i].width);
+    sim_stream->height = static_cast<uint16_t>(streams[i].height);
+    sim_stream->minBitrate = streams[i].min_bitrate_bps / 1000;
+    sim_stream->targetBitrate = streams[i].target_bitrate_bps / 1000;
+    sim_stream->maxBitrate = streams[i].max_bitrate_bps / 1000;
+    sim_stream->qpMax = streams[i].max_qp;
+    sim_stream->numberOfTemporalLayers = static_cast<unsigned char>(
+        streams[i].temporal_layer_thresholds_bps.size() + 1);
+
+    video_codec.width =
+        std::max(video_codec.width, static_cast<uint16_t>(streams[i].width));
+    video_codec.height =
+        std::max(video_codec.height, static_cast<uint16_t>(streams[i].height));
+    video_codec.minBitrate =
+        std::min(static_cast<uint16_t>(video_codec.minBitrate),
+                 static_cast<uint16_t>(streams[i].min_bitrate_bps / 1000));
+    video_codec.maxBitrate += streams[i].max_bitrate_bps / 1000;
+    video_codec.qpMax = std::max(video_codec.qpMax,
+                                 static_cast<unsigned int>(streams[i].max_qp));
+  }
+
+  if (video_codec.maxBitrate == 0) {
+    // Unset max bitrate -> cap to one bit per pixel.
+    video_codec.maxBitrate =
+        (video_codec.width * video_codec.height * video_codec.maxFramerate) /
+        1000;
+  }
+  if (video_codec.maxBitrate < kEncoderMinBitrateKbps)
+    video_codec.maxBitrate = kEncoderMinBitrateKbps;
+
+  RTC_DCHECK_GT(streams[0].max_framerate, 0);
+  video_codec.maxFramerate = streams[0].max_framerate;
+  video_codec.expect_encode_from_texture = config.expect_encode_from_texture;
+
+  return video_codec;
+}
+
+// TODO(pbos): Lower these thresholds (to closer to 100%) when we handle
+// pipelining encoders better (multiple input frames before something comes
+// out). This should effectively turn off CPU adaptations for systems that
+// remotely cope with the load right now.
+CpuOveruseOptions GetCpuOveruseOptions(bool full_overuse_time) {
+  CpuOveruseOptions options;
+  if (full_overuse_time) {
+    options.low_encode_usage_threshold_percent = 150;
+    options.high_encode_usage_threshold_percent = 200;
+  }
+  return options;
+}
+
+}  //  namespace
+
+class ViEEncoder::EncodeTask : public rtc::QueuedTask {
+ public:
+  EncodeTask(const VideoFrame& frame, ViEEncoder* vie_encoder)
+      : vie_encoder_(vie_encoder) {
+    frame_.ShallowCopy(frame);
+    ++vie_encoder_->posted_frames_waiting_for_encode_;
+  }
+
+ private:
+  bool Run() override {
+    RTC_DCHECK_GT(vie_encoder_->posted_frames_waiting_for_encode_.Value(), 0);
+    if (--vie_encoder_->posted_frames_waiting_for_encode_ == 0) {
+      vie_encoder_->EncodeVideoFrame(frame_);
+    } else {
+      // There is a newer frame in flight. Do not encode this frame.
+      LOG(LS_VERBOSE)
+          << "Incoming frame dropped due to that the encoder is blocked.";
+    }
+    return true;
+  }
+  VideoFrame frame_;
+  ViEEncoder* vie_encoder_;
+};
+
 ViEEncoder::ViEEncoder(uint32_t number_of_cores,
-                       ProcessThread* module_process_thread,
                        SendStatisticsProxy* stats_proxy,
-                       OveruseFrameDetector* overuse_detector,
-                       EncodedImageCallback* sink)
-    : number_of_cores_(number_of_cores),
-      sink_(sink),
+                       const VideoSendStream::Config::EncoderSettings& settings,
+                       rtc::VideoSinkInterface<VideoFrame>* pre_encode_callback,
+                       LoadObserver* overuse_callback,
+                       EncodedFrameObserver* encoder_timing)
+    : shutdown_event_(true /* manual_reset */, false),
+      number_of_cores_(number_of_cores),
+      settings_(settings),
       vp_(VideoProcessing::Create()),
       video_sender_(Clock::GetRealTimeClock(), this, this),
+      overuse_detector_(Clock::GetRealTimeClock(),
+                        GetCpuOveruseOptions(settings.full_overuse_time),
+                        this,
+                        encoder_timing,
+                        stats_proxy),
+      load_observer_(overuse_callback),
       stats_proxy_(stats_proxy),
-      overuse_detector_(overuse_detector),
-      time_of_last_frame_activity_ms_(std::numeric_limits<int64_t>::max()),
+      pre_encode_callback_(pre_encode_callback),
+      module_process_thread_(nullptr),
       encoder_config_(),
+      encoder_start_bitrate_bps_(0),
       last_observed_bitrate_bps_(0),
       encoder_paused_and_dropped_frame_(false),
-      module_process_thread_(module_process_thread),
       has_received_sli_(false),
       picture_id_sli_(0),
       has_received_rpsi_(false),
       picture_id_rpsi_(0),
-      video_suspended_(false) {
+      clock_(Clock::GetRealTimeClock()),
+      last_captured_timestamp_(0),
+      delta_ntp_internal_ms_(clock_->CurrentNtpInMilliseconds() -
+                             clock_->TimeInMilliseconds()),
+      encoder_queue_("EncoderQueue") {
+  vp_->EnableTemporalDecimation(false);
+  encoded_sequenced_checker_.Detach();
+
+  encoder_queue_.PostTask([this] {
+    RTC_DCHECK_RUN_ON(&encoder_queue_);
+    video_sender_.RegisterExternalEncoder(
+        settings_.encoder, settings_.payload_type, settings_.internal_source);
+  });
+}
+
+ViEEncoder::~ViEEncoder() {
+  RTC_DCHECK(shutdown_event_.Wait(0))
+      << "Must call ::Stop() before destruction.";
+}
+
+void ViEEncoder::Stop() {
+  if (!encoder_queue_.IsCurrent()) {
+    encoder_queue_.PostTask([this] { Stop(); });
+    shutdown_event_.Wait(rtc::Event::kForever);
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
+  video_sender_.RegisterExternalEncoder(nullptr, settings_.payload_type, false);
+  shutdown_event_.Set();
+}
+
+void ViEEncoder::RegisterProcessThread(ProcessThread* module_process_thread) {
+  RTC_DCHECK(!module_process_thread_);
+  module_process_thread_ = module_process_thread;
+  module_process_thread_->RegisterModule(&overuse_detector_);
   module_process_thread_->RegisterModule(&video_sender_);
-  vp_->EnableTemporalDecimation(true);
-}
-
-vcm::VideoSender* ViEEncoder::video_sender() {
-  return &video_sender_;
-}
-
-ViEEncoder::~ViEEncoder() {
+  module_process_thread_checker_.DetachFromThread();
+}
+
+void ViEEncoder::DeRegisterProcessThread() {
+  module_process_thread_->DeRegisterModule(&overuse_detector_);
   module_process_thread_->DeRegisterModule(&video_sender_);
 }
 
-int32_t ViEEncoder::RegisterExternalEncoder(webrtc::VideoEncoder* encoder,
-                                            uint8_t pl_type,
-                                            bool internal_source) {
-  video_sender_.RegisterExternalEncoder(encoder, pl_type, internal_source);
-  return 0;
-}
-
-int32_t ViEEncoder::DeRegisterExternalEncoder(uint8_t pl_type) {
-  video_sender_.RegisterExternalEncoder(nullptr, pl_type, false);
-  return 0;
-}
-
-void ViEEncoder::SetEncoder(const webrtc::VideoCodec& video_codec,
-                            size_t max_data_payload_length) {
+void ViEEncoder::SetSink(EncodedImageCallback* sink) {
+  encoder_queue_.PostTask([this, sink] {
+    RTC_DCHECK_RUN_ON(&encoder_queue_);
+    sink_ = sink;
+  });
+}
+
+void ViEEncoder::SetStartBitrate(int start_bitrate_bps) {
+  encoder_queue_.PostTask([this, start_bitrate_bps] {
+    RTC_DCHECK_RUN_ON(&encoder_queue_);
+    encoder_start_bitrate_bps_ = start_bitrate_bps;
+  });
+}
+
+void ViEEncoder::ConfigureEncoder(const VideoEncoderConfig& config,
+                                  size_t max_data_payload_length) {
+  VideoCodec video_codec = VideoEncoderConfigToVideoCodec(
+      config, settings_.payload_name, settings_.payload_type);
+  encoder_queue_.PostTask([this, video_codec, max_data_payload_length] {
+    ConfigureEncoderInternal(video_codec, max_data_payload_length);
+  });
+  return;
+}
+
+void ViEEncoder::ConfigureEncoderInternal(const VideoCodec& video_codec,
+                                          size_t max_data_payload_length) {
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
+  RTC_DCHECK_GE(encoder_start_bitrate_bps_, 0);
+  RTC_DCHECK(sink_);
+
   // Setting target width and height for VPM.
   RTC_CHECK_EQ(VPM_OK,
                vp_->SetTargetResolution(video_codec.width, video_codec.height,
                                         video_codec.maxFramerate));
-  {
-    rtc::CritScope lock(&data_cs_);
-    encoder_config_ = video_codec;
-  }
+
+  encoder_config_ = video_codec;
+  encoder_config_.startBitrate = encoder_start_bitrate_bps_ / 1000;
 
   bool success = video_sender_.RegisterSendCodec(
-                     &video_codec, number_of_cores_,
+                     &encoder_config_, number_of_cores_,
                      static_cast<uint32_t>(max_data_payload_length)) == VCM_OK;
 
   if (!success) {
     LOG(LS_ERROR) << "Failed to configure encoder.";
     RTC_DCHECK(success);
   }
 
   if (stats_proxy_) {
     VideoEncoderConfig::ContentType content_type =
         VideoEncoderConfig::ContentType::kRealtimeVideo;
     switch (video_codec.mode) {
       case kRealtimeVideo:
         content_type = VideoEncoderConfig::ContentType::kRealtimeVideo;
         break;
       case kScreensharing:
         content_type = VideoEncoderConfig::ContentType::kScreen;
         break;
       default:
         RTC_NOTREACHED();
         break;
     }
     stats_proxy_->SetContentType(content_type);
   }
 }
 
+void ViEEncoder::IncomingCapturedFrame(const VideoFrame& video_frame) {
+  RTC_DCHECK_RUNS_SERIALIZED(&incoming_frame_race_checker_);
+  stats_proxy_->OnIncomingFrame(video_frame.width(), video_frame.height());
+
+  VideoFrame incoming_frame = video_frame;
+
+  // Local time in webrtc time base.
+  int64_t current_time = clock_->TimeInMilliseconds();
+  incoming_frame.set_render_time_ms(current_time);
+
+  // Capture time may come from clock with an offset and drift from clock_.
+  int64_t capture_ntp_time_ms;
+  if (video_frame.ntp_time_ms() != 0) {
+    capture_ntp_time_ms = video_frame.ntp_time_ms();
+  } else if (video_frame.render_time_ms() != 0) {
+    capture_ntp_time_ms = video_frame.render_time_ms() + delta_ntp_internal_ms_;
+  } else {
+    capture_ntp_time_ms = current_time + delta_ntp_internal_ms_;
+  }
+  incoming_frame.set_ntp_time_ms(capture_ntp_time_ms);
+
+  // Convert NTP time, in ms, to RTP timestamp.
+  const int kMsToRtpTimestamp = 90;
+  incoming_frame.set_timestamp(
+      kMsToRtpTimestamp * static_cast<uint32_t>(incoming_frame.ntp_time_ms()));
+
+  if (incoming_frame.ntp_time_ms() <= last_captured_timestamp_) {
+    // We don't allow the same capture time for two frames, drop this one.
+    LOG(LS_WARNING) << "Same/old NTP timestamp ("
+                    << incoming_frame.ntp_time_ms()
+                    << " <= " << last_captured_timestamp_
+                    << ") for incoming frame. Dropping.";
+    return;
+  }
+
+  last_captured_timestamp_ = incoming_frame.ntp_time_ms();
+  overuse_detector_.FrameCaptured(incoming_frame);
+  encoder_queue_.PostTask(
+      std::unique_ptr<rtc::QueuedTask>(new EncodeTask(incoming_frame, this)));
+}
+
 bool ViEEncoder::EncoderPaused() const {
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
   // Pause video if paused by caller or as long as the network is down or the
   // pacer queue has grown too large in buffered mode.
   // If the pacer queue has grown too large or the network is down,
   // last_observed_bitrate_bps_ will be 0.
-  return video_suspended_ || last_observed_bitrate_bps_ == 0;
+  return last_observed_bitrate_bps_ == 0;
 }
 
 void ViEEncoder::TraceFrameDropStart() {
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
   // Start trace event only on the first frame after encoder is paused.
   if (!encoder_paused_and_dropped_frame_) {
     TRACE_EVENT_ASYNC_BEGIN0("webrtc", "EncoderPaused", this);
   }
   encoder_paused_and_dropped_frame_ = true;
   return;
 }
 
 void ViEEncoder::TraceFrameDropEnd() {
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
   // End trace event on first frame after encoder resumes, if frame was dropped.
   if (encoder_paused_and_dropped_frame_) {
     TRACE_EVENT_ASYNC_END0("webrtc", "EncoderPaused", this);
   }
   encoder_paused_and_dropped_frame_ = false;
 }
 
 void ViEEncoder::EncodeVideoFrame(const VideoFrame& video_frame) {
-  VideoCodecType codec_type;
-  {
-    rtc::CritScope lock(&data_cs_);
-    time_of_last_frame_activity_ms_ = rtc::TimeMillis();
-    if (EncoderPaused()) {
-      TraceFrameDropStart();
-      return;
-    }
-    TraceFrameDropEnd();
-    codec_type = encoder_config_.codecType;
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
+  if (pre_encode_callback_)
+    pre_encode_callback_->OnFrame(video_frame);
+
+  if (EncoderPaused()) {
+    TraceFrameDropStart();
+    return;
   }
+  TraceFrameDropEnd();
 
   TRACE_EVENT_ASYNC_STEP0("webrtc", "Video", video_frame.render_time_ms(),
                           "Encode");
   const VideoFrame* frame_to_send = &video_frame;
   // TODO(wuchengli): support texture frames.
   if (!video_frame.video_frame_buffer()->native_handle()) {
     // Pass frame via preprocessor.
     frame_to_send = vp_->PreprocessFrame(video_frame);
     if (!frame_to_send) {
       // Drop this frame, or there was an error processing it.
       return;
     }
   }
 
-  if (codec_type == webrtc::kVideoCodecVP8) {
+  if (encoder_config_.codecType == webrtc::kVideoCodecVP8) {
     webrtc::CodecSpecificInfo codec_specific_info;
     codec_specific_info.codecType = webrtc::kVideoCodecVP8;
-    {
-      rtc::CritScope lock(&data_cs_);
+
       codec_specific_info.codecSpecific.VP8.hasReceivedRPSI =
           has_received_rpsi_;
       codec_specific_info.codecSpecific.VP8.hasReceivedSLI =
           has_received_sli_;
       codec_specific_info.codecSpecific.VP8.pictureIdRPSI =
           picture_id_rpsi_;
       codec_specific_info.codecSpecific.VP8.pictureIdSLI  =
           picture_id_sli_;
       has_received_sli_ = false;
       has_received_rpsi_ = false;
-    }
 
     video_sender_.AddVideoFrame(*frame_to_send, &codec_specific_info);
     return;
   }
   video_sender_.AddVideoFrame(*frame_to_send, nullptr);
 }
 
 void ViEEncoder::SendKeyFrame() {
+  if (!encoder_queue_.IsCurrent()) {
+    encoder_queue_.PostTask([this] { SendKeyFrame(); });
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
   video_sender_.IntraFrameRequest(0);
 }
 
-int64_t ViEEncoder::time_of_last_frame_activity_ms() {
-  rtc::CritScope lock(&data_cs_);
-  return time_of_last_frame_activity_ms_;
-}
-
 int32_t ViEEncoder::Encoded(const EncodedImage& encoded_image,
                             const CodecSpecificInfo* codec_specific_info,
                             const RTPFragmentationHeader* fragmentation) {
-  {
-    rtc::CritScope lock(&data_cs_);
-    time_of_last_frame_activity_ms_ = rtc::TimeMillis();
-  }
+  // Encoded is called on whatever thread the real encoder implementation run
+  // on.
+  RTC_DCHECK(encoded_sequenced_checker_.CalledSequentially());

[pbos-webrtc, 2016/07/13 12:35:39]

This can be called in parallel for simulcast encoders based on hardware
encoders.

I believe this DCHECK is invalid, though technically you could have one race
checker per simulcast layer (though not a sequential task checker).

[perkj_webrtc, 2016/07/14 10:11:28]

ok, thanks. updated the comment.

   if (stats_proxy_) {
     stats_proxy_->OnSendEncodedImage(encoded_image, codec_specific_info);
   }
 
   int success =
       sink_->Encoded(encoded_image, codec_specific_info, fragmentation);
 
-  overuse_detector_->FrameSent(encoded_image._timeStamp);
+  overuse_detector_.FrameSent(encoded_image._timeStamp);
   return success;
 }
 
 void ViEEncoder::SendStatistics(uint32_t bit_rate,
                                 uint32_t frame_rate,
                                 const std::string& encoder_name) {
+  RTC_DCHECK(module_process_thread_checker_.CalledOnValidThread());
   if (stats_proxy_)
     stats_proxy_->OnEncoderStatsUpdate(frame_rate, bit_rate, encoder_name);
 }
 
 void ViEEncoder::OnReceivedSLI(uint8_t picture_id) {
-  rtc::CritScope lock(&data_cs_);
+  if (!encoder_queue_.IsCurrent()) {
+    encoder_queue_.PostTask([this, picture_id] { OnReceivedSLI(picture_id); });
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
   picture_id_sli_ = picture_id;
   has_received_sli_ = true;
 }
 
 void ViEEncoder::OnReceivedRPSI(uint64_t picture_id) {
-  rtc::CritScope lock(&data_cs_);
+  if (!encoder_queue_.IsCurrent()) {
+    encoder_queue_.PostTask([this, picture_id] { OnReceivedRPSI(picture_id); });
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
   picture_id_rpsi_ = picture_id;
   has_received_rpsi_ = true;
 }
 
 void ViEEncoder::OnReceivedIntraFrameRequest(size_t stream_index) {
+  if (!encoder_queue_.IsCurrent()) {
+    encoder_queue_.PostTask(
+        [this, stream_index] { OnReceivedIntraFrameRequest(stream_index); });
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
   // Key frame request from remote side, signal to VCM.
   TRACE_EVENT0("webrtc", "OnKeyFrameRequest");
   video_sender_.IntraFrameRequest(stream_index);
 }
 
 void ViEEncoder::OnBitrateUpdated(uint32_t bitrate_bps,
                                   uint8_t fraction_lost,
                                   int64_t round_trip_time_ms) {
+  if (!encoder_queue_.IsCurrent()) {
+    encoder_queue_.PostTask(
+        [this, bitrate_bps, fraction_lost, round_trip_time_ms] {
+          OnBitrateUpdated(bitrate_bps, fraction_lost, round_trip_time_ms);
+        });
+    return;
+  }
+  RTC_DCHECK_RUN_ON(&encoder_queue_);
+  RTC_DCHECK(sink_) << "sink_ must be set before the encoder is active.";
+
   LOG(LS_VERBOSE) << "OnBitrateUpdated, bitrate " << bitrate_bps
                   << " packet loss " << static_cast<int>(fraction_lost)
                   << " rtt " << round_trip_time_ms;
+
   video_sender_.SetChannelParameters(bitrate_bps, fraction_lost,
                                      round_trip_time_ms);
-  bool video_suspension_changed;
+
+  encoder_start_bitrate_bps_ =
+      bitrate_bps != 0 ? bitrate_bps : encoder_start_bitrate_bps_;
   bool video_is_suspended = bitrate_bps == 0;
-  {
-    rtc::CritScope lock(&data_cs_);
-    last_observed_bitrate_bps_ = bitrate_bps;
-    video_suspension_changed = video_suspended_ != video_is_suspended;
-    video_suspended_ = video_is_suspended;
-    // Set |time_of_last_frame_activity_ms_| to now if this is the first time
-    // the encoder is supposed to produce encoded frames.
-    // TODO(perkj): Remove this hack. It is here to avoid a race that the
-    // encoder report that it has timed out before it has processed the first
-    // frame.
-    if (last_observed_bitrate_bps_ != 0 &&
-        time_of_last_frame_activity_ms_ ==
-            std::numeric_limits<int64_t>::max()) {
-      time_of_last_frame_activity_ms_ = rtc::TimeMillis();
-    }
-  }
+  bool video_suspension_changed =
+      video_is_suspended != (last_observed_bitrate_bps_ == 0);
+  last_observed_bitrate_bps_ = bitrate_bps;
 
   if (stats_proxy_ && video_suspension_changed) {
     LOG(LS_INFO) << "Video suspend state changed to: "
                  << (video_is_suspended ? "suspended" : "not suspended");
     stats_proxy_->OnSuspendChange(video_is_suspended);
   }
 }
 
+void ViEEncoder::OveruseDetected() {
+  RTC_DCHECK_RUN_ON(&module_process_thread_checker_);
+  // TODO(perkj): When ViEEncoder inherit rtc::VideoSink instead of
+  // VideoCaptureInput |load_observer_| should be removed and overuse be
+  // expressed as rtc::VideoSinkWants instead.
+  if (load_observer_)
+    load_observer_->OnLoadUpdate(LoadObserver::kOveruse);
+}
+
+void ViEEncoder::NormalUsage() {
+  RTC_DCHECK_RUN_ON(&module_process_thread_checker_);
+  if (load_observer_)
+    load_observer_->OnLoadUpdate(LoadObserver::kUnderuse);
+}
+
 }  // namespace webrtc