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) {

[stefan-webrtc, 2016/07/08 15:56:42]

I think a lot of the changes in this file should be moved to a separate CL to
reduce the complexity of this CL. It will be way easier to review in pieces...
Would be good if as much as possible of the stuff that doesn't directly involved
the task queue was done in isolated CLs. A huge CL might become difficult to
revert later if we find out it's causing some problem.

I'm mostly thinking of these methods and the moving of timestamping, which all
seem pretty isolated.

[perkj_webrtc, 2016/07/11 11:41:08]

I have spend a couple of weeks to try to land smaller changes but always found
one dependency that made a test break. In the end I gave up and did this huge cl
at once. If you feel strong about it I can try again with this part as you
suggest. I am afraid it will require a bit of changes in VideoSendStream that
would be discarded in the next cl. Before this cl, the encoder thread lived in
VideoSendStream.

+  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_thread_checker_.DetachFromThread();
+
+  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_);

[stefan-webrtc, 2016/07/08 15:56:42]

Is there a point in DCHECKing this? Seems like it's basically verifying that the
encoder queue works at all, which I guess is already verified by its unittests?

[perkj_webrtc, 2016/07/11 11:41:08]

Needed to compile. Otherwise AccessOn thread annotation gets angry.

+    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,

[stefan-webrtc, 2016/07/08 15:56:42]

Most methods in this class don't have a corresponding "Internal" version. Maybe
you should merge this one with ConfigureEncoder above too?

[perkj_webrtc, 2016/07/11 11:41:08]

I can't due to  VideoCodec video_codec = VideoEncoderConfigToVideoCodec(...

+                                          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_);
+  // TODO(perkj): Timestamp handling is moved here from the removed
+  // implementation class VideoCaptureInput. nisse@ is looking into the
+  // timestamps in https://bugs.chromium.org/p/webrtc/issues/detail?id=5740.

[stefan-webrtc, 2016/07/08 15:56:42]

It's not clear what the TODO is. How do we determine that this is TODO is done
and can be removed?

[perkj_webrtc, 2016/07/11 11:41:08]

I removed the whole todo in that case. It was added here as a notice that this
code need some love. We have 3 timestamps in the video frame and the below logic
is not clear to me why it is done the way it is but I definetly don't want to
change it now.

+  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_thread_checker_.CalledOnValidThread());
   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