xref: /external/webrtc/webrtc/call/congestion_controller.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_engine/vie_channel_group.h"

#include "webrtc/base/checks.h"
#include "webrtc/base/thread_annotations.h"
#include "webrtc/common.h"
#include "webrtc/modules/pacing/include/paced_sender.h"
#include "webrtc/modules/pacing/include/packet_router.h"
#include "webrtc/modules/remote_bitrate_estimator/include/send_time_history.h"
#include "webrtc/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h"
#include "webrtc/modules/remote_bitrate_estimator/remote_bitrate_estimator_single_stream.h"
#include "webrtc/modules/remote_bitrate_estimator/remote_estimator_proxy.h"
#include "webrtc/modules/remote_bitrate_estimator/transport_feedback_adapter.h"
#include "webrtc/modules/rtp_rtcp/interface/rtp_rtcp.h"
#include "webrtc/modules/utility/interface/process_thread.h"
#include "webrtc/system_wrappers/interface/critical_section_wrapper.h"
#include "webrtc/system_wrappers/interface/logging.h"
#include "webrtc/video_engine/call_stats.h"
#include "webrtc/video_engine/encoder_state_feedback.h"
#include "webrtc/video_engine/payload_router.h"
#include "webrtc/video_engine/vie_channel.h"
#include "webrtc/video_engine/vie_encoder.h"
#include "webrtc/video_engine/vie_remb.h"
#include "webrtc/voice_engine/include/voe_video_sync.h"

namespace webrtc {
namespace {

static const uint32_t kTimeOffsetSwitchThreshold = 30;

class WrappingBitrateEstimator : public RemoteBitrateEstimator {
 public:
  WrappingBitrateEstimator(RemoteBitrateObserver* observer, Clock* clock)
      : observer_(observer),
        clock_(clock),
        crit_sect_(CriticalSectionWrapper::CreateCriticalSection()),
        rbe_(new RemoteBitrateEstimatorSingleStream(observer_, clock_)),
        using_absolute_send_time_(false),
        packets_since_absolute_send_time_(0),
        min_bitrate_bps_(RemoteBitrateEstimator::kDefaultMinBitrateBps) {}

  virtual ~WrappingBitrateEstimator() {}

  void IncomingPacket(int64_t arrival_time_ms,
                      size_t payload_size,
                      const RTPHeader& header,
                      bool was_paced) override {
    CriticalSectionScoped cs(crit_sect_.get());
    PickEstimatorFromHeader(header);
    rbe_->IncomingPacket(arrival_time_ms, payload_size, header, was_paced);
  }

  int32_t Process() override {
    CriticalSectionScoped cs(crit_sect_.get());
    return rbe_->Process();
  }

  int64_t TimeUntilNextProcess() override {
    CriticalSectionScoped cs(crit_sect_.get());
    return rbe_->TimeUntilNextProcess();
  }

  void OnRttUpdate(int64_t avg_rtt_ms, int64_t max_rtt_ms) override {
    CriticalSectionScoped cs(crit_sect_.get());
    rbe_->OnRttUpdate(avg_rtt_ms, max_rtt_ms);
  }

  void RemoveStream(unsigned int ssrc) override {
    CriticalSectionScoped cs(crit_sect_.get());
    rbe_->RemoveStream(ssrc);
  }

  bool LatestEstimate(std::vector<unsigned int>* ssrcs,
                      unsigned int* bitrate_bps) const override {
    CriticalSectionScoped cs(crit_sect_.get());
    return rbe_->LatestEstimate(ssrcs, bitrate_bps);
  }

  bool GetStats(ReceiveBandwidthEstimatorStats* output) const override {
    CriticalSectionScoped cs(crit_sect_.get());
    return rbe_->GetStats(output);
  }

  void SetMinBitrate(int min_bitrate_bps) {
    CriticalSectionScoped cs(crit_sect_.get());
    rbe_->SetMinBitrate(min_bitrate_bps);
    min_bitrate_bps_ = min_bitrate_bps;
  }

 private:
  void PickEstimatorFromHeader(const RTPHeader& header)
      EXCLUSIVE_LOCKS_REQUIRED(crit_sect_.get()) {
    if (header.extension.hasAbsoluteSendTime) {
      // If we see AST in header, switch RBE strategy immediately.
      if (!using_absolute_send_time_) {
        LOG(LS_INFO) <<
            "WrappingBitrateEstimator: Switching to absolute send time RBE.";
        using_absolute_send_time_ = true;
        PickEstimator();
      }
      packets_since_absolute_send_time_ = 0;
    } else {
      // When we don't see AST, wait for a few packets before going back to TOF.
      if (using_absolute_send_time_) {
        ++packets_since_absolute_send_time_;
        if (packets_since_absolute_send_time_ >= kTimeOffsetSwitchThreshold) {
          LOG(LS_INFO) << "WrappingBitrateEstimator: Switching to transmission "
                       << "time offset RBE.";
          using_absolute_send_time_ = false;
          PickEstimator();
        }
      }
    }
  }

  // Instantiate RBE for Time Offset or Absolute Send Time extensions.
  void PickEstimator() EXCLUSIVE_LOCKS_REQUIRED(crit_sect_.get()) {
    if (using_absolute_send_time_) {
      rbe_.reset(new RemoteBitrateEstimatorAbsSendTime(observer_, clock_));
    } else {
      rbe_.reset(new RemoteBitrateEstimatorSingleStream(observer_, clock_));
    }
    rbe_->SetMinBitrate(min_bitrate_bps_);
  }

  RemoteBitrateObserver* observer_;
  Clock* clock_;
  rtc::scoped_ptr<CriticalSectionWrapper> crit_sect_;
  rtc::scoped_ptr<RemoteBitrateEstimator> rbe_;
  bool using_absolute_send_time_;
  uint32_t packets_since_absolute_send_time_;
  int min_bitrate_bps_;

  RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(WrappingBitrateEstimator);
};

}  // namespace

ChannelGroup::ChannelGroup(ProcessThread* process_thread)
    : remb_(new VieRemb()),
      bitrate_allocator_(new BitrateAllocator()),
      call_stats_(new CallStats()),
      packet_router_(new PacketRouter()),
      pacer_(new PacedSender(Clock::GetRealTimeClock(),
                             packet_router_.get(),
                             BitrateController::kDefaultStartBitrateKbps,
                             PacedSender::kDefaultPaceMultiplier *
                                 BitrateController::kDefaultStartBitrateKbps,
                             0)),
      remote_bitrate_estimator_(
          new WrappingBitrateEstimator(remb_.get(), Clock::GetRealTimeClock())),
      remote_estimator_proxy_(
          new RemoteEstimatorProxy(Clock::GetRealTimeClock(),
                                   packet_router_.get())),
      encoder_state_feedback_(new EncoderStateFeedback()),
      process_thread_(process_thread),
      pacer_thread_(ProcessThread::Create("PacerThread")),
      // Constructed last as this object calls the provided callback on
      // construction.
      bitrate_controller_(
          BitrateController::CreateBitrateController(Clock::GetRealTimeClock(),
                                                     this)),
      min_bitrate_bps_(RemoteBitrateEstimator::kDefaultMinBitrateBps) {
  call_stats_->RegisterStatsObserver(remote_bitrate_estimator_.get());

  pacer_thread_->RegisterModule(pacer_.get());
  pacer_thread_->Start();

  process_thread->RegisterModule(remote_estimator_proxy_.get());
  process_thread->RegisterModule(remote_bitrate_estimator_.get());
  process_thread->RegisterModule(call_stats_.get());
  process_thread->RegisterModule(bitrate_controller_.get());
}

ChannelGroup::~ChannelGroup() {
  pacer_thread_->Stop();
  pacer_thread_->DeRegisterModule(pacer_.get());
  process_thread_->DeRegisterModule(bitrate_controller_.get());
  process_thread_->DeRegisterModule(call_stats_.get());
  process_thread_->DeRegisterModule(remote_bitrate_estimator_.get());
  process_thread_->DeRegisterModule(remote_estimator_proxy_.get());
  call_stats_->DeregisterStatsObserver(remote_bitrate_estimator_.get());
  if (transport_feedback_adapter_.get())
    call_stats_->DeregisterStatsObserver(transport_feedback_adapter_.get());
  RTC_DCHECK(channel_map_.empty());
  RTC_DCHECK(!remb_->InUse());
  RTC_DCHECK(vie_encoder_map_.empty());
}

bool ChannelGroup::CreateSendChannel(int channel_id,
                                     Transport* transport,
                                     SendStatisticsProxy* stats_proxy,
                                     I420FrameCallback* pre_encode_callback,
                                     int number_of_cores,
                                     const VideoSendStream::Config& config) {
  TransportFeedbackObserver* transport_feedback_observer = nullptr;
  bool transport_seq_enabled = false;
  for (const RtpExtension& extension : config.rtp.extensions) {
    if (extension.name == RtpExtension::kTransportSequenceNumber) {
      transport_seq_enabled = true;
      break;
    }
  }
  if (transport_seq_enabled) {
    if (transport_feedback_adapter_.get() == nullptr) {
      transport_feedback_adapter_.reset(new TransportFeedbackAdapter(
          bitrate_controller_->CreateRtcpBandwidthObserver(),
          Clock::GetRealTimeClock(), process_thread_));
      transport_feedback_adapter_->SetBitrateEstimator(
          new RemoteBitrateEstimatorAbsSendTime(
              transport_feedback_adapter_.get(), Clock::GetRealTimeClock()));
      transport_feedback_adapter_->GetBitrateEstimator()->SetMinBitrate(
          min_bitrate_bps_);
      call_stats_->RegisterStatsObserver(transport_feedback_adapter_.get());
    }
    transport_feedback_observer = transport_feedback_adapter_.get();
  }

  const std::vector<uint32_t>& ssrcs = config.rtp.ssrcs;
  RTC_DCHECK(!ssrcs.empty());
  rtc::scoped_ptr<ViEEncoder> vie_encoder(new ViEEncoder(
      channel_id, number_of_cores, process_thread_, stats_proxy,
      pre_encode_callback, pacer_.get(), bitrate_allocator_.get()));
  if (!vie_encoder->Init()) {
    return false;
  }
  ViEEncoder* encoder = vie_encoder.get();
  if (!CreateChannel(channel_id, transport, number_of_cores,
                     vie_encoder.release(), ssrcs.size(), true,
                     remote_bitrate_estimator_.get(),
                     transport_feedback_observer)) {
    return false;
  }
  ViEChannel* channel = channel_map_[channel_id];
  // Connect the encoder with the send packet router, to enable sending.
  encoder->StartThreadsAndSetSharedMembers(channel->send_payload_router(),
                                           channel->vcm_protection_callback());

  encoder_state_feedback_->AddEncoder(ssrcs, encoder);
  std::vector<uint32_t> first_ssrc(1, ssrcs[0]);
  encoder->SetSsrcs(first_ssrc);
  return true;
}

bool ChannelGroup::CreateReceiveChannel(
    int channel_id,
    Transport* transport,
    int number_of_cores,
    const VideoReceiveStream::Config& config) {
  bool send_side_bwe = false;
  for (const RtpExtension& extension : config.rtp.extensions) {
    if (extension.name == RtpExtension::kTransportSequenceNumber) {
      send_side_bwe = true;
      break;
    }
  }

  RemoteBitrateEstimator* bitrate_estimator;
  if (send_side_bwe) {
    bitrate_estimator = remote_estimator_proxy_.get();
  } else {
    bitrate_estimator = remote_bitrate_estimator_.get();
  }
  return CreateChannel(channel_id, transport, number_of_cores, nullptr, 1,
                       false, bitrate_estimator, nullptr);
}

bool ChannelGroup::CreateChannel(int channel_id,
                                 Transport* transport,
                                 int number_of_cores,
                                 ViEEncoder* vie_encoder,
                                 size_t max_rtp_streams,
                                 bool sender,
                                 RemoteBitrateEstimator* bitrate_estimator,
                                 TransportFeedbackObserver* feedback_observer) {
  rtc::scoped_ptr<ViEChannel> channel(new ViEChannel(
      number_of_cores, transport, process_thread_,
      encoder_state_feedback_->GetRtcpIntraFrameObserver(),
      bitrate_controller_->CreateRtcpBandwidthObserver(), feedback_observer,
      bitrate_estimator, call_stats_->rtcp_rtt_stats(), pacer_.get(),
      packet_router_.get(), max_rtp_streams, sender));
  if (channel->Init() != 0) {
    return false;
  }

  // Register the channel to receive stats updates.
  call_stats_->RegisterStatsObserver(channel->GetStatsObserver());

  // Store the channel, add it to the channel group and save the vie_encoder.
  channel_map_[channel_id] = channel.release();
  if (vie_encoder) {
    rtc::CritScope lock(&encoder_map_crit_);
    vie_encoder_map_[channel_id] = vie_encoder;
  }

  return true;
}

void ChannelGroup::DeleteChannel(int channel_id) {
  ViEChannel* vie_channel = PopChannel(channel_id);

  ViEEncoder* vie_encoder = GetEncoder(channel_id);

  call_stats_->DeregisterStatsObserver(vie_channel->GetStatsObserver());
  SetChannelRembStatus(false, false, vie_channel);

  // If we're a sender, remove the feedback and stop all encoding threads and
  // processing. This must be done before deleting the channel.
  if (vie_encoder) {
    encoder_state_feedback_->RemoveEncoder(vie_encoder);
    vie_encoder->StopThreadsAndRemoveSharedMembers();
  }

  unsigned int remote_ssrc = 0;
  vie_channel->GetRemoteSSRC(&remote_ssrc);
  channel_map_.erase(channel_id);
  remote_bitrate_estimator_->RemoveStream(remote_ssrc);

  delete vie_channel;

  if (vie_encoder) {
    {
      rtc::CritScope lock(&encoder_map_crit_);
      vie_encoder_map_.erase(vie_encoder_map_.find(channel_id));
    }
    delete vie_encoder;
  }

  LOG(LS_VERBOSE) << "Channel deleted " << channel_id;
}

ViEChannel* ChannelGroup::GetChannel(int channel_id) const {
  ChannelMap::const_iterator it = channel_map_.find(channel_id);
  if (it == channel_map_.end()) {
    LOG(LS_ERROR) << "Channel doesn't exist " << channel_id;
    return NULL;
  }
  return it->second;
}

ViEEncoder* ChannelGroup::GetEncoder(int channel_id) const {
  rtc::CritScope lock(&encoder_map_crit_);
  EncoderMap::const_iterator it = vie_encoder_map_.find(channel_id);
  if (it == vie_encoder_map_.end())
    return nullptr;
  return it->second;
}

ViEChannel* ChannelGroup::PopChannel(int channel_id) {
  ChannelMap::iterator c_it = channel_map_.find(channel_id);
  RTC_DCHECK(c_it != channel_map_.end());
  ViEChannel* channel = c_it->second;
  channel_map_.erase(c_it);

  return channel;
}

void ChannelGroup::SetSyncInterface(VoEVideoSync* sync_interface) {
  for (auto channel : channel_map_)
    channel.second->SetVoiceChannel(-1, sync_interface);
}

void ChannelGroup::SetBweBitrates(int min_bitrate_bps,
                                  int start_bitrate_bps,
                                  int max_bitrate_bps) {
  if (start_bitrate_bps > 0)
    bitrate_controller_->SetStartBitrate(start_bitrate_bps);
  bitrate_controller_->SetMinMaxBitrate(min_bitrate_bps, max_bitrate_bps);
  if (remote_bitrate_estimator_.get())
    remote_bitrate_estimator_->SetMinBitrate(min_bitrate_bps);
  if (transport_feedback_adapter_.get())
    transport_feedback_adapter_->GetBitrateEstimator()->SetMinBitrate(
        min_bitrate_bps);
  min_bitrate_bps_ = min_bitrate_bps;
}

BitrateController* ChannelGroup::GetBitrateController() const {
  return bitrate_controller_.get();
}

RemoteBitrateEstimator* ChannelGroup::GetRemoteBitrateEstimator() const {
  return remote_bitrate_estimator_.get();
}

CallStats* ChannelGroup::GetCallStats() const {
  return call_stats_.get();
}

EncoderStateFeedback* ChannelGroup::GetEncoderStateFeedback() const {
  return encoder_state_feedback_.get();
}

int64_t ChannelGroup::GetPacerQueuingDelayMs() const {
  return pacer_->QueueInMs();
}

void ChannelGroup::SetChannelRembStatus(bool sender,
                                        bool receiver,
                                        ViEChannel* channel) {
  // Update the channel state.
  channel->EnableRemb(sender || receiver);
  // Update the REMB instance with necessary RTP modules.
  RtpRtcp* rtp_module = channel->rtp_rtcp();
  if (sender) {
    remb_->AddRembSender(rtp_module);
  } else {
    remb_->RemoveRembSender(rtp_module);
  }
  if (receiver) {
    remb_->AddReceiveChannel(rtp_module);
  } else {
    remb_->RemoveReceiveChannel(rtp_module);
  }
}

void ChannelGroup::OnNetworkChanged(uint32_t target_bitrate_bps,
                                    uint8_t fraction_loss,
                                    int64_t rtt) {
  bitrate_allocator_->OnNetworkChanged(target_bitrate_bps, fraction_loss, rtt);
  int pad_up_to_bitrate_bps = 0;
  {
    rtc::CritScope lock(&encoder_map_crit_);
    for (const auto& encoder : vie_encoder_map_)
      pad_up_to_bitrate_bps += encoder.second->GetPaddingNeededBps();
  }
  pacer_->UpdateBitrate(
      target_bitrate_bps / 1000,
      PacedSender::kDefaultPaceMultiplier * target_bitrate_bps / 1000,
      pad_up_to_bitrate_bps / 1000);
}
}  // namespace webrtc