media / cast / logging / encoding_event_subscriber.cc [blame]

// Copyright 2014 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "media/cast/logging/encoding_event_subscriber.h"

#include <stdint.h>

#include <algorithm>
#include <cstring>
#include <utility>

#include "base/logging.h"
#include "base/not_fatal_until.h"
#include "base/numerics/safe_conversions.h"
#include "media/cast/logging/proto/proto_utils.h"

using google::protobuf::RepeatedPtrField;
using media::cast::proto::AggregatedFrameEvent;
using media::cast::proto::AggregatedPacketEvent;
using media::cast::proto::BasePacketEvent;
using media::cast::proto::LogMetadata;

namespace {

// A size limit on maps to keep lookups fast.
const size_t kMaxMapSize = 200;

// The smallest (oredered by RTP timestamp) |kNumMapEntriesToTransfer| entries
// will be moved when the map size reaches |kMaxMapSize|.
// Must be smaller than |kMaxMapSize|.
const size_t kNumMapEntriesToTransfer = 100;

template <typename ProtoPtr>
bool IsRtpTimestampLessThan(const ProtoPtr& lhs, const ProtoPtr& rhs) {
  return lhs->relative_rtp_timestamp() < rhs->relative_rtp_timestamp();
}

BasePacketEvent* GetNewBasePacketEvent(AggregatedPacketEvent* event_proto,
                                       int packet_id,
                                       int size) {
  BasePacketEvent* base = event_proto->add_base_packet_event();
  base->set_packet_id(packet_id);
  base->set_size(size);
  return base;
}

}  // namespace

namespace media {
namespace cast {

EncodingEventSubscriber::EncodingEventSubscriber(
    EventMediaType event_media_type,
    size_t max_frames)
    : event_media_type_(event_media_type), max_frames_(max_frames) {
  Reset();
}

EncodingEventSubscriber::~EncodingEventSubscriber() {
  DCHECK(thread_checker_.CalledOnValidThread());
}

void EncodingEventSubscriber::OnReceiveFrameEvent(
    const FrameEvent& frame_event) {
  DCHECK(thread_checker_.CalledOnValidThread());

  if (event_media_type_ != frame_event.media_type) {
    return;
  }

  const RtpTimeDelta relative_rtp_timestamp =
      GetRelativeRtpTimestamp(frame_event.rtp_timestamp);
  const uint32_t lower_32_bits = relative_rtp_timestamp.lower_32_bits();
  AggregatedFrameEvent* event_proto_ptr = nullptr;

  // Look up existing entry. If not found, create a new entry and add to map.
  auto it = frame_event_map_.find(relative_rtp_timestamp);
  if (it == frame_event_map_.end()) {
    if (!ShouldCreateNewProto(lower_32_bits)) {
      return;
    }

    IncrementStoredProtoCount(lower_32_bits);
    auto event_proto = std::make_unique<AggregatedFrameEvent>();
    event_proto->set_relative_rtp_timestamp(lower_32_bits);
    event_proto_ptr = event_proto.get();
    frame_event_map_.insert(
        std::make_pair(relative_rtp_timestamp, std::move(event_proto)));
  } else {
    if (it->second->event_type_size() >= kMaxEventsPerProto) {
      DVLOG(2) << "Too many events in frame " << frame_event.rtp_timestamp
               << ". Using new frame event proto.";
      AddFrameEventToStorage(std::move(it->second));
      if (!ShouldCreateNewProto(lower_32_bits)) {
        frame_event_map_.erase(it);
        return;
      }

      IncrementStoredProtoCount(lower_32_bits);
      it->second = std::make_unique<AggregatedFrameEvent>();
      it->second->set_relative_rtp_timestamp(lower_32_bits);
    }
    event_proto_ptr = it->second.get();
  }

  event_proto_ptr->add_event_type(ToProtoEventType(frame_event.type));
  event_proto_ptr->add_event_timestamp_ms(
      (frame_event.timestamp - base::TimeTicks()).InMilliseconds());

  if (frame_event.type == FRAME_CAPTURE_END) {
    if (frame_event.media_type == VIDEO_EVENT && frame_event.width > 0 &&
        frame_event.height > 0) {
      event_proto_ptr->set_width(frame_event.width);
      event_proto_ptr->set_height(frame_event.height);
    }
  } else if (frame_event.type == FRAME_ENCODED) {
    event_proto_ptr->set_encoded_frame_size(frame_event.size);
    if (frame_event.encoder_cpu_utilization >= 0.0) {
      event_proto_ptr->set_encoder_cpu_percent_utilized(
          base::saturated_cast<int32_t>(
              frame_event.encoder_cpu_utilization * 100.0 + 0.5));
    }
    if (frame_event.idealized_bitrate_utilization >= 0.0) {
      event_proto_ptr->set_idealized_bitrate_percent_utilized(
          base::saturated_cast<int32_t>(
              frame_event.idealized_bitrate_utilization * 100.0 + 0.5));
    }
    if (frame_event.media_type == VIDEO_EVENT) {
      event_proto_ptr->set_key_frame(frame_event.key_frame);
      event_proto_ptr->set_target_bitrate(frame_event.target_bitrate);
    }
  } else if (frame_event.type == FRAME_PLAYOUT) {
    event_proto_ptr->set_delay_millis(frame_event.delay_delta.InMilliseconds());
  }

  if (frame_event_map_.size() > kMaxMapSize) {
    TransferFrameEvents(kNumMapEntriesToTransfer);
  }

  DCHECK(frame_event_map_.size() <= kMaxMapSize);
  DCHECK(frame_event_storage_.size() <= max_frames_);
}

void EncodingEventSubscriber::OnReceivePacketEvent(
    const PacketEvent& packet_event) {
  DCHECK(thread_checker_.CalledOnValidThread());

  if (event_media_type_ != packet_event.media_type) {
    return;
  }

  const RtpTimeDelta relative_rtp_timestamp =
      GetRelativeRtpTimestamp(packet_event.rtp_timestamp);
  uint32_t lower_32_bits = relative_rtp_timestamp.lower_32_bits();
  auto it = packet_event_map_.find(relative_rtp_timestamp);
  BasePacketEvent* base_packet_event_proto = nullptr;

  // Look up existing entry. If not found, create a new entry and add to map.
  if (it == packet_event_map_.end()) {
    if (!ShouldCreateNewProto(lower_32_bits)) {
      return;
    }

    IncrementStoredProtoCount(lower_32_bits);
    auto event_proto = std::make_unique<AggregatedPacketEvent>();
    event_proto->set_relative_rtp_timestamp(lower_32_bits);
    base_packet_event_proto = GetNewBasePacketEvent(
        event_proto.get(), packet_event.packet_id, packet_event.size);
    packet_event_map_.insert(
        std::make_pair(relative_rtp_timestamp, std::move(event_proto)));
  } else {
    // Found existing entry, now look up existing BasePacketEvent using packet
    // ID. If not found, create a new entry and add to proto.
    RepeatedPtrField<BasePacketEvent>* field =
        it->second->mutable_base_packet_event();
    for (RepeatedPtrField<BasePacketEvent>::pointer_iterator base_it =
             field->pointer_begin();
         base_it != field->pointer_end(); ++base_it) {
      if ((*base_it)->packet_id() == packet_event.packet_id) {
        base_packet_event_proto = *base_it;
        break;
      }
    }
    if (!base_packet_event_proto) {
      if (it->second->base_packet_event_size() >= kMaxPacketsPerFrame) {
        DVLOG(3) << "Too many packets in AggregatedPacketEvent "
                 << packet_event.rtp_timestamp << ". "
                 << "Using new packet event proto.";
        AddPacketEventToStorage(std::move(it->second));
        if (!ShouldCreateNewProto(lower_32_bits)) {
          packet_event_map_.erase(it);
          return;
        }

        IncrementStoredProtoCount(lower_32_bits);
        it->second = std::make_unique<AggregatedPacketEvent>();
        it->second->set_relative_rtp_timestamp(lower_32_bits);
      }

      base_packet_event_proto = GetNewBasePacketEvent(
          it->second.get(), packet_event.packet_id, packet_event.size);
    } else if (base_packet_event_proto->event_type_size() >=
               kMaxEventsPerProto) {
      DVLOG(3) << "Too many events in packet " << packet_event.rtp_timestamp
               << ", " << packet_event.packet_id
               << ". Using new packet event proto.";
      AddPacketEventToStorage(std::move(it->second));
      if (!ShouldCreateNewProto(lower_32_bits)) {
        packet_event_map_.erase(it);
        return;
      }

      IncrementStoredProtoCount(lower_32_bits);
      it->second = std::make_unique<AggregatedPacketEvent>();
      it->second->set_relative_rtp_timestamp(lower_32_bits);
      base_packet_event_proto = GetNewBasePacketEvent(
          it->second.get(), packet_event.packet_id, packet_event.size);
    }
  }

  base_packet_event_proto->add_event_type(ToProtoEventType(packet_event.type));
  base_packet_event_proto->add_event_timestamp_ms(
      (packet_event.timestamp - base::TimeTicks()).InMilliseconds());

  // |base_packet_event_proto| could have been created with a receiver event
  // which does not have the packet size and we would need to overwrite it when
  // we see a sender event, which does have the packet size.
  if (packet_event.size > 0) {
    base_packet_event_proto->set_size(packet_event.size);
  }

  if (packet_event_map_.size() > kMaxMapSize) {
    TransferPacketEvents(kNumMapEntriesToTransfer);
  }

  DCHECK(packet_event_map_.size() <= kMaxMapSize);
  DCHECK(packet_event_storage_.size() <= max_frames_);
}

void EncodingEventSubscriber::GetEventsAndReset(
    LogMetadata* metadata,
    FrameEventList* frame_events,
    PacketEventList* packet_events) {
  DCHECK(thread_checker_.CalledOnValidThread());

  // Flush all events.
  TransferFrameEvents(frame_event_map_.size());
  TransferPacketEvents(packet_event_map_.size());
  std::sort(frame_event_storage_.begin(), frame_event_storage_.end(),
            &IsRtpTimestampLessThan<std::unique_ptr<AggregatedFrameEvent>>);
  std::sort(packet_event_storage_.begin(), packet_event_storage_.end(),
            &IsRtpTimestampLessThan<std::unique_ptr<AggregatedPacketEvent>>);

  metadata->set_is_audio(event_media_type_ == AUDIO_EVENT);
  metadata->set_first_rtp_timestamp(first_rtp_timestamp_.lower_32_bits());
  metadata->set_num_frame_events(frame_event_storage_.size());
  metadata->set_num_packet_events(packet_event_storage_.size());
  metadata->set_reference_timestamp_ms_at_unix_epoch(
      (base::TimeTicks::UnixEpoch() - base::TimeTicks()).InMilliseconds());
  frame_events->swap(frame_event_storage_);
  packet_events->swap(packet_event_storage_);
  Reset();
}

void EncodingEventSubscriber::TransferFrameEvents(size_t max_num_entries) {
  DCHECK(frame_event_map_.size() >= max_num_entries);

  auto it = frame_event_map_.begin();
  for (size_t i = 0; i < max_num_entries && it != frame_event_map_.end();
       i++, ++it) {
    AddFrameEventToStorage(std::move(it->second));
  }

  frame_event_map_.erase(frame_event_map_.begin(), it);
}

void EncodingEventSubscriber::TransferPacketEvents(size_t max_num_entries) {
  auto it = packet_event_map_.begin();
  for (size_t i = 0; i < max_num_entries && it != packet_event_map_.end();
       i++, ++it) {
    AddPacketEventToStorage(std::move(it->second));
  }

  packet_event_map_.erase(packet_event_map_.begin(), it);
}

void EncodingEventSubscriber::AddFrameEventToStorage(
    std::unique_ptr<AggregatedFrameEvent> frame_event_proto) {
  if (frame_event_storage_.size() >= max_frames_) {
    auto& entry = frame_event_storage_[frame_event_storage_index_];
    DecrementStoredProtoCount(entry->relative_rtp_timestamp());
    entry = std::move(frame_event_proto);
  } else {
    frame_event_storage_.push_back(std::move(frame_event_proto));
  }

  frame_event_storage_index_ = (frame_event_storage_index_ + 1) % max_frames_;
}

void EncodingEventSubscriber::AddPacketEventToStorage(
    std::unique_ptr<AggregatedPacketEvent> packet_event_proto) {
  if (packet_event_storage_.size() >= max_frames_) {
    auto& entry = packet_event_storage_[packet_event_storage_index_];
    DecrementStoredProtoCount(entry->relative_rtp_timestamp());
    entry = std::move(packet_event_proto);
  } else {
    packet_event_storage_.push_back(std::move(packet_event_proto));
  }

  packet_event_storage_index_ = (packet_event_storage_index_ + 1) % max_frames_;
}

bool EncodingEventSubscriber::ShouldCreateNewProto(
    uint32_t relative_rtp_timestamp_lower_32_bits) const {
  auto it = stored_proto_counts_.find(relative_rtp_timestamp_lower_32_bits);
  int proto_count = it == stored_proto_counts_.end() ? 0 : it->second;
  DVLOG_IF(2, proto_count >= kMaxProtosPerFrame)
      << relative_rtp_timestamp_lower_32_bits
      << " already reached max number of protos.";
  return proto_count < kMaxProtosPerFrame;
}

void EncodingEventSubscriber::IncrementStoredProtoCount(
    uint32_t relative_rtp_timestamp_lower_32_bits) {
  stored_proto_counts_[relative_rtp_timestamp_lower_32_bits]++;
  DCHECK_LE(stored_proto_counts_[relative_rtp_timestamp_lower_32_bits],
            kMaxProtosPerFrame)
      << relative_rtp_timestamp_lower_32_bits
      << " exceeded max number of event protos.";
}

void EncodingEventSubscriber::DecrementStoredProtoCount(
    uint32_t relative_rtp_timestamp_lower_32_bits) {
  auto it = stored_proto_counts_.find(relative_rtp_timestamp_lower_32_bits);
  CHECK(it != stored_proto_counts_.end(), base::NotFatalUntil::M130)
      << "no event protos for " << relative_rtp_timestamp_lower_32_bits;
  if (it->second > 1) {
    it->second--;
  } else {
    stored_proto_counts_.erase(it);
  }
}

RtpTimeDelta EncodingEventSubscriber::GetRelativeRtpTimestamp(
    RtpTimeTicks rtp_timestamp) {
  if (!seen_first_rtp_timestamp_) {
    seen_first_rtp_timestamp_ = true;
    first_rtp_timestamp_ = rtp_timestamp;
  }

  return rtp_timestamp - first_rtp_timestamp_;
}

void EncodingEventSubscriber::Reset() {
  frame_event_map_.clear();
  frame_event_storage_.clear();
  frame_event_storage_index_ = 0;
  packet_event_map_.clear();
  packet_event_storage_.clear();
  packet_event_storage_index_ = 0;
  stored_proto_counts_.clear();
  seen_first_rtp_timestamp_ = false;
  first_rtp_timestamp_ = RtpTimeTicks();
}

}  // namespace cast
}  // namespace media