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