media/cast/logging/encoding_event_subscriber.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "media/cast/logging/encoding_event_subscriber.h"
   6
   7 #include <algorithm>
   8 #include <cstring>
   9 #include <utility>
  10
  11 #include "base/logging.h"
  12 #include "media/cast/logging/proto/proto_utils.h"
  13
  14 using google::protobuf::RepeatedPtrField;
  15 using media::cast::proto::AggregatedFrameEvent;
  16 using media::cast::proto::AggregatedPacketEvent;
  17 using media::cast::proto::BasePacketEvent;
  18 using media::cast::proto::LogMetadata;
  19
  20 namespace {
  21
  22 // A size limit on maps to keep lookups fast.
  23 const size_t kMaxMapSize = 200;
  24
  25 // The smallest (oredered by RTP timestamp) |kNumMapEntriesToTransfer| entries
  26 // will be moved when the map size reaches |kMaxMapSize|.
  27 // Must be smaller than |kMaxMapSize|.
  28 const size_t kNumMapEntriesToTransfer = 100;
  29
  30 template <typename ProtoPtr>
  31 bool IsRtpTimestampLessThan(const ProtoPtr& lhs, const ProtoPtr& rhs) {
  32   return lhs->relative_rtp_timestamp() < rhs->relative_rtp_timestamp();
  33 }
  34
  35 BasePacketEvent* GetNewBasePacketEvent(AggregatedPacketEvent* event_proto,
  36     int packet_id, int size) {
  37   BasePacketEvent* base = event_proto->add_base_packet_event();
  38   base->set_packet_id(packet_id);
  39   base->set_size(size);
  40   return base;
  41 }
  42
  43 }
  44
  45 namespace media {
  46 namespace cast {
  47
  48 EncodingEventSubscriber::EncodingEventSubscriber(
  49     EventMediaType event_media_type,
  50     size_t max_frames)
  51     : event_media_type_(event_media_type),
  52       max_frames_(max_frames),
  53       frame_event_storage_index_(0),
  54       packet_event_storage_index_(0),
  55       seen_first_rtp_timestamp_(false),
  56       first_rtp_timestamp_(0u) {}
  57
  58 EncodingEventSubscriber::~EncodingEventSubscriber() {
  59   DCHECK(thread_checker_.CalledOnValidThread());
  60 }
  61
  62 void EncodingEventSubscriber::OnReceiveFrameEvent(
  63     const FrameEvent& frame_event) {
  64   DCHECK(thread_checker_.CalledOnValidThread());
  65
  66   if (event_media_type_ != frame_event.media_type)
  67     return;
  68
  69   RtpTimestamp relative_rtp_timestamp =
  70       GetRelativeRtpTimestamp(frame_event.rtp_timestamp);
  71   FrameEventMap::iterator it = frame_event_map_.find(relative_rtp_timestamp);
  72   linked_ptr<AggregatedFrameEvent> event_proto;
  73
  74   // Look up existing entry. If not found, create a new entry and add to map.
  75   if (it == frame_event_map_.end()) {
  76     event_proto.reset(new AggregatedFrameEvent);
  77     event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
  78     frame_event_map_.insert(
  79         std::make_pair(relative_rtp_timestamp, event_proto));
  80   } else {
  81     event_proto = it->second;
  82     if (event_proto->event_type_size() >= kMaxEventsPerProto) {
  83       DVLOG(2) << "Too many events in frame " << frame_event.rtp_timestamp
  84                << ". Using new frame event proto.";
  85       AddFrameEventToStorage(event_proto);
  86       event_proto.reset(new AggregatedFrameEvent);
  87       event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
  88       it->second = event_proto;
  89     }
  90   }
  91
  92   event_proto->add_event_type(ToProtoEventType(frame_event.type));
  93   event_proto->add_event_timestamp_ms(
  94       (frame_event.timestamp - base::TimeTicks()).InMilliseconds());
  95
  96   if (frame_event.type == FRAME_ENCODED) {
  97     event_proto->set_encoded_frame_size(frame_event.size);
  98     if (frame_event.media_type == VIDEO_EVENT) {
  99       event_proto->set_encoded_frame_size(frame_event.size);
 100       event_proto->set_key_frame(frame_event.key_frame);
 101       event_proto->set_target_bitrate(frame_event.target_bitrate);
 102     }
 103   } else if (frame_event.type == FRAME_PLAYOUT) {
 104     event_proto->set_delay_millis(frame_event.delay_delta.InMilliseconds());
 105   }
 106
 107   if (frame_event_map_.size() > kMaxMapSize)
 108     TransferFrameEvents(kNumMapEntriesToTransfer);
 109
 110   DCHECK(frame_event_map_.size() <= kMaxMapSize);
 111   DCHECK(frame_event_storage_.size() <= max_frames_);
 112 }
 113
 114 void EncodingEventSubscriber::OnReceivePacketEvent(
 115     const PacketEvent& packet_event) {
 116   DCHECK(thread_checker_.CalledOnValidThread());
 117
 118   if (event_media_type_ != packet_event.media_type)
 119     return;
 120
 121   RtpTimestamp relative_rtp_timestamp =
 122       GetRelativeRtpTimestamp(packet_event.rtp_timestamp);
 123   PacketEventMap::iterator it =
 124       packet_event_map_.find(relative_rtp_timestamp);
 125   linked_ptr<AggregatedPacketEvent> event_proto;
 126   BasePacketEvent* base_packet_event_proto = NULL;
 127
 128   // Look up existing entry. If not found, create a new entry and add to map.
 129   if (it == packet_event_map_.end()) {
 130     event_proto.reset(new AggregatedPacketEvent);
 131     event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
 132     packet_event_map_.insert(
 133         std::make_pair(relative_rtp_timestamp, event_proto));
 134     base_packet_event_proto = GetNewBasePacketEvent(
 135         event_proto.get(), packet_event.packet_id, packet_event.size);
 136   } else {
 137     // Found existing entry, now look up existing BasePacketEvent using packet
 138     // ID. If not found, create a new entry and add to proto.
 139     event_proto = it->second;
 140     RepeatedPtrField<BasePacketEvent>* field =
 141         event_proto->mutable_base_packet_event();
 142     for (RepeatedPtrField<BasePacketEvent>::pointer_iterator base_it =
 143              field->pointer_begin();
 144          base_it != field->pointer_end();
 145          ++base_it) {
 146       if ((*base_it)->packet_id() == packet_event.packet_id) {
 147         base_packet_event_proto = *base_it;
 148         break;
 149       }
 150     }
 151     if (!base_packet_event_proto) {
 152       if (event_proto->base_packet_event_size() >= kMaxPacketsPerFrame) {
 153         DVLOG(3) << "Too many packets in AggregatedPacketEvent "
 154                  << packet_event.rtp_timestamp << ". "
 155                  << "Using new packet event proto.";
 156         AddPacketEventToStorage(event_proto);
 157         event_proto.reset(new AggregatedPacketEvent);
 158         event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
 159         it->second = event_proto;
 160       }
 161
 162       base_packet_event_proto = GetNewBasePacketEvent(
 163           event_proto.get(), packet_event.packet_id, packet_event.size);
 164     } else if (base_packet_event_proto->event_type_size() >=
 165                kMaxEventsPerProto) {
 166       DVLOG(3) << "Too many events in packet "
 167                << packet_event.rtp_timestamp << ", "
 168                << packet_event.packet_id << ". Using new packet event proto.";
 169       AddPacketEventToStorage(event_proto);
 170       event_proto.reset(new AggregatedPacketEvent);
 171       event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
 172       it->second = event_proto;
 173       base_packet_event_proto = GetNewBasePacketEvent(
 174           event_proto.get(), packet_event.packet_id, packet_event.size);
 175     }
 176   }
 177
 178   base_packet_event_proto->add_event_type(
 179       ToProtoEventType(packet_event.type));
 180   base_packet_event_proto->add_event_timestamp_ms(
 181       (packet_event.timestamp - base::TimeTicks()).InMilliseconds());
 182
 183   // |base_packet_event_proto| could have been created with a receiver event
 184   // which does not have the packet size and we would need to overwrite it when
 185   // we see a sender event, which does have the packet size.
 186   if (packet_event.size > 0) {
 187     base_packet_event_proto->set_size(packet_event.size);
 188   }
 189
 190   if (packet_event_map_.size() > kMaxMapSize)
 191     TransferPacketEvents(kNumMapEntriesToTransfer);
 192
 193   DCHECK(packet_event_map_.size() <= kMaxMapSize);
 194   DCHECK(packet_event_storage_.size() <= max_frames_);
 195 }
 196
 197 void EncodingEventSubscriber::GetEventsAndReset(LogMetadata* metadata,
 198     FrameEventList* frame_events, PacketEventList* packet_events) {
 199   DCHECK(thread_checker_.CalledOnValidThread());
 200
 201   // Flush all events.
 202   TransferFrameEvents(frame_event_map_.size());
 203   TransferPacketEvents(packet_event_map_.size());
 204   std::sort(frame_event_storage_.begin(), frame_event_storage_.end(),
 205             &IsRtpTimestampLessThan<linked_ptr<AggregatedFrameEvent> >);
 206   std::sort(packet_event_storage_.begin(), packet_event_storage_.end(),
 207             &IsRtpTimestampLessThan<linked_ptr<AggregatedPacketEvent> >);
 208
 209   metadata->set_is_audio(event_media_type_ == AUDIO_EVENT);
 210   metadata->set_first_rtp_timestamp(first_rtp_timestamp_);
 211   metadata->set_num_frame_events(frame_event_storage_.size());
 212   metadata->set_num_packet_events(packet_event_storage_.size());
 213   metadata->set_reference_timestamp_ms_at_unix_epoch(
 214       (base::TimeTicks::UnixEpoch() - base::TimeTicks()).InMilliseconds());
 215   frame_events->swap(frame_event_storage_);
 216   packet_events->swap(packet_event_storage_);
 217   Reset();
 218 }
 219
 220 void EncodingEventSubscriber::TransferFrameEvents(size_t max_num_entries) {
 221   DCHECK(frame_event_map_.size() >= max_num_entries);
 222
 223   FrameEventMap::iterator it = frame_event_map_.begin();
 224   for (size_t i = 0;
 225        i < max_num_entries && it != frame_event_map_.end();
 226        i++, ++it) {
 227     AddFrameEventToStorage(it->second);
 228   }
 229
 230   frame_event_map_.erase(frame_event_map_.begin(), it);
 231 }
 232
 233 void EncodingEventSubscriber::TransferPacketEvents(size_t max_num_entries) {
 234   PacketEventMap::iterator it = packet_event_map_.begin();
 235   for (size_t i = 0;
 236        i < max_num_entries && it != packet_event_map_.end();
 237        i++, ++it) {
 238     AddPacketEventToStorage(it->second);
 239   }
 240
 241   packet_event_map_.erase(packet_event_map_.begin(), it);
 242 }
 243
 244 void EncodingEventSubscriber::AddFrameEventToStorage(
 245     const linked_ptr<AggregatedFrameEvent>& frame_event_proto) {
 246   if (frame_event_storage_.size() >= max_frames_) {
 247     frame_event_storage_[frame_event_storage_index_] = frame_event_proto;
 248   } else {
 249     frame_event_storage_.push_back(frame_event_proto);
 250   }
 251
 252   frame_event_storage_index_ = (frame_event_storage_index_ + 1) % max_frames_;
 253 }
 254
 255 void EncodingEventSubscriber::AddPacketEventToStorage(
 256     const linked_ptr<AggregatedPacketEvent>& packet_event_proto) {
 257   if (packet_event_storage_.size() >= max_frames_)
 258     packet_event_storage_[packet_event_storage_index_] = packet_event_proto;
 259   else
 260     packet_event_storage_.push_back(packet_event_proto);
 261
 262   packet_event_storage_index_ = (packet_event_storage_index_ + 1) % max_frames_;
 263 }
 264
 265 RtpTimestamp EncodingEventSubscriber::GetRelativeRtpTimestamp(
 266     RtpTimestamp rtp_timestamp) {
 267   if (!seen_first_rtp_timestamp_) {
 268     seen_first_rtp_timestamp_ = true;
 269     first_rtp_timestamp_ = rtp_timestamp;
 270   }
 271
 272   return rtp_timestamp - first_rtp_timestamp_;
 273 }
 274
 275 void EncodingEventSubscriber::Reset() {
 276   frame_event_map_.clear();
 277   frame_event_storage_.clear();
 278   frame_event_storage_index_ = 0;
 279   packet_event_map_.clear();
 280   packet_event_storage_.clear();
 281   packet_event_storage_index_ = 0;
 282   seen_first_rtp_timestamp_ = false;
 283   first_rtp_timestamp_ = 0u;
 284 }
 285
 286 }  // namespace cast
 287 }  // namespace media