Cast: Implemented size limiting and event filtering in EncodingEventSubscriber.

media/cast/logging/encoding_event_subscriber.cc

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // 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 <cstring>
 #include <utility>
 
 #include "base/logging.h"
 #include "media/cast/logging/proto/proto_utils.h"
 
 using google::protobuf::RepeatedPtrField;
 using media::cast::proto::AggregatedFrameEvent;
 using media::cast::proto::AggregatedGenericEvent;
 using media::cast::proto::AggregatedPacketEvent;
 using media::cast::proto::BasePacketEvent;
 
 namespace media {
 namespace cast {
 
-EncodingEventSubscriber::EncodingEventSubscriber() {}
+EncodingEventSubscriber::EncodingEventSubscriber(
+    EncodingEventSubscriberConfig config)
+    : config_(config) {
+  DCHECK_GT(config_.max_generic_event_values, 0);
+  memset(next_write_index_, 0, sizeof(next_write_index_));
+}
 
 EncodingEventSubscriber::~EncodingEventSubscriber() {
   DCHECK(thread_checker_.CalledOnValidThread());
 }
 
 void EncodingEventSubscriber::OnReceiveFrameEvent(
     const FrameEvent& frame_event) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  FrameEventMap::iterator it = frame_event_map_.find(frame_event.rtp_timestamp);
-  linked_ptr<AggregatedFrameEvent> event_proto;
+  if (ShouldProcessEvent(frame_event.type)) {
+    FrameEventMap::iterator it =
+        frame_event_map_.find(frame_event.rtp_timestamp);
+    linked_ptr<AggregatedFrameEvent> event_proto;
 
-  // Look up existing entry. If not found, create a new entry and add to map.
-  if (it == frame_event_map_.end()) {
-    event_proto.reset(new AggregatedFrameEvent);
-    event_proto->set_rtp_timestamp(frame_event.rtp_timestamp);
-    frame_event_map_.insert(
-        std::make_pair(frame_event.rtp_timestamp, event_proto));
-  } else {
-    event_proto = it->second;
+    // Look up existing entry. If not found, create a new entry and add to map.
+    if (it == frame_event_map_.end()) {
+      event_proto.reset(new AggregatedFrameEvent);
+      event_proto->set_rtp_timestamp(frame_event.rtp_timestamp);
+      frame_event_map_.insert(
+          std::make_pair(frame_event.rtp_timestamp, event_proto));
+    } else {
+      event_proto = it->second;
+    }
+
+    event_proto->add_event_type(ToProtoEventType(frame_event.type));
+    event_proto->add_event_timestamp_micros(
+        frame_event.timestamp.ToInternalValue());
+
+    if (frame_event.type == kAudioFrameEncoded ||
+        frame_event.type == kVideoFrameEncoded) {
+      event_proto->set_encoded_frame_size(frame_event.size);
+    } else if (frame_event.type == kAudioPlayoutDelay ||
+               frame_event.type == kVideoRenderDelay) {
+      event_proto->set_delay_millis(frame_event.delay_delta.InMilliseconds());
+    }
+
+    TruncateFrameEventMapIfNeeded();
   }
 
-  event_proto->add_event_type(ToProtoEventType(frame_event.type));
-  event_proto->add_event_timestamp_micros(
-      frame_event.timestamp.ToInternalValue());
-
-  if (frame_event.type == kAudioFrameEncoded ||
-      frame_event.type == kVideoFrameEncoded) {
-    event_proto->set_encoded_frame_size(frame_event.size);
-  } else if (frame_event.type == kAudioPlayoutDelay ||
-             frame_event.type == kVideoRenderDelay) {
-    event_proto->set_delay_millis(frame_event.delay_delta.InMilliseconds());
-  }
+  DCHECK(frame_event_map_.size() <= config_.max_frames);
 }
 
 void EncodingEventSubscriber::OnReceivePacketEvent(
     const PacketEvent& packet_event) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  PacketEventMap::iterator it =
-      packet_event_map_.find(packet_event.rtp_timestamp);
-  linked_ptr<AggregatedPacketEvent> event_proto;
-  BasePacketEvent* base_packet_event_proto = NULL;
+  if (ShouldProcessEvent(packet_event.type)) {
+    PacketEventMap::iterator it =
+        packet_event_map_.find(packet_event.rtp_timestamp);
+    linked_ptr<AggregatedPacketEvent> event_proto;
+    BasePacketEvent* base_packet_event_proto = NULL;
 
-  // Look up existing entry. If not found, create a new entry and add to map.
-  if (it == packet_event_map_.end()) {
-    event_proto.reset(new AggregatedPacketEvent);
-    event_proto->set_rtp_timestamp(packet_event.rtp_timestamp);
-    packet_event_map_.insert(
-        std::make_pair(packet_event.rtp_timestamp, event_proto));
-    base_packet_event_proto = event_proto->add_base_packet_event();
-    base_packet_event_proto->set_packet_id(packet_event.packet_id);
-  } else {
-    // Found existing entry, now look up existing BasePacketEvent using packet
-    // ID. If not found, create a new entry and add to proto.
-    event_proto = it->second;
-    RepeatedPtrField<BasePacketEvent>* field =
-        event_proto->mutable_base_packet_event();
-    for (RepeatedPtrField<BasePacketEvent>::pointer_iterator it =
-             field->pointer_begin();
-         it != field->pointer_end(); ++it) {
-      if ((*it)->packet_id() == packet_event.packet_id) {
-        base_packet_event_proto = *it;
-        break;
+    // Look up existing entry. If not found, create a new entry and add to map.
+    if (it == packet_event_map_.end()) {
+      event_proto.reset(new AggregatedPacketEvent);
+      event_proto->set_rtp_timestamp(packet_event.rtp_timestamp);
+      packet_event_map_.insert(
+          std::make_pair(packet_event.rtp_timestamp, event_proto));
+      base_packet_event_proto = event_proto->add_base_packet_event();
+      base_packet_event_proto->set_packet_id(packet_event.packet_id);
+    } else {
+      // Found existing entry, now look up existing BasePacketEvent using packet
+      // ID. If not found, create a new entry and add to proto.
+      event_proto = it->second;
+      RepeatedPtrField<BasePacketEvent>* field =
+          event_proto->mutable_base_packet_event();
+      for (RepeatedPtrField<BasePacketEvent>::pointer_iterator it =
+               field->pointer_begin();
+           it != field->pointer_end();
+           ++it) {
+        if ((*it)->packet_id() == packet_event.packet_id) {
+          base_packet_event_proto = *it;
+          break;
+        }
+      }
+      if (!base_packet_event_proto) {
+        base_packet_event_proto = event_proto->add_base_packet_event();
+        base_packet_event_proto->set_packet_id(packet_event.packet_id);
       }
     }
-    if (!base_packet_event_proto) {
-      base_packet_event_proto = event_proto->add_base_packet_event();
-      base_packet_event_proto->set_packet_id(packet_event.packet_id);
-    }
+
+    base_packet_event_proto->add_event_type(
+        ToProtoEventType(packet_event.type));
+    base_packet_event_proto->add_event_timestamp_micros(
+        packet_event.timestamp.ToInternalValue());
+
+    TruncatePacketEventMapIfNeeded();
   }
 
-  base_packet_event_proto->add_event_type(ToProtoEventType(packet_event.type));
-  base_packet_event_proto->add_event_timestamp_micros(
-      packet_event.timestamp.ToInternalValue());
+  DCHECK(packet_event_map_.size() <= config_.max_frames);
 }
 
 void EncodingEventSubscriber::OnReceiveGenericEvent(

[Alpha Left Google, 2014/02/14 18:42:39]

What are the generic events that we capture? I'm not sure we care about these
events.

[imcheng, 2014/02/14 20:24:17]

As mentioned in the other file, there are 7 generic events. There are 4 that it
sounds like we are not interested in, and the other 3 shouldn't be generic
events in the first place.

     const GenericEvent& generic_event) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  GenericEventMap::iterator it = generic_event_map_.find(generic_event.type);
-  linked_ptr<AggregatedGenericEvent> event_proto;
-  if (it == generic_event_map_.end()) {
-    event_proto.reset(new AggregatedGenericEvent);
-    event_proto->set_event_type(ToProtoEventType(generic_event.type));
-    generic_event_map_.insert(std::make_pair(generic_event.type, event_proto));
-  } else {
-    event_proto = it->second;
+  if (ShouldProcessEvent(generic_event.type)) {
+    GenericEventMap::iterator it = generic_event_map_.find(generic_event.type);
+    linked_ptr<AggregatedGenericEvent> event_proto;
+    if (it == generic_event_map_.end()) {
+      event_proto.reset(new AggregatedGenericEvent);
+      event_proto->set_event_type(ToProtoEventType(generic_event.type));
+      generic_event_map_.insert(
+          std::make_pair(generic_event.type, event_proto));
+    } else {
+      event_proto = it->second;
+    }
+
+    if (event_proto->value_size() < config_.max_generic_event_values) {
+      event_proto->add_event_timestamp_micros(
+          generic_event.timestamp.ToInternalValue());
+      event_proto->add_value(generic_event.value);
+    } else {
+      // Reached maximum size, treat the repeated fields as circular buffer and
+      // rewrite oldest value.
+      int index = next_write_index_[generic_event.type];
+      event_proto->set_event_timestamp_micros(
+          index, generic_event.timestamp.ToInternalValue());
+      event_proto->set_value(index, generic_event.value);
+      next_write_index_[generic_event.type] =
+          (index + 1) % config_.max_generic_event_values;
+    }
   }
-
-  event_proto->add_event_timestamp_micros(
-      generic_event.timestamp.ToInternalValue());
-  event_proto->add_value(generic_event.value);
 }
 
 void EncodingEventSubscriber::GetFrameEventsAndReset(
     FrameEventMap* frame_event_map) {
   DCHECK(thread_checker_.CalledOnValidThread());
   frame_event_map->swap(frame_event_map_);
   frame_event_map_.clear();
 }
 
 void EncodingEventSubscriber::GetPacketEventsAndReset(
     PacketEventMap* packet_event_map) {
   DCHECK(thread_checker_.CalledOnValidThread());
   packet_event_map->swap(packet_event_map_);
   packet_event_map_.clear();
 }
 
 void EncodingEventSubscriber::GetGenericEventsAndReset(
     GenericEventMap* generic_event_map) {
   DCHECK(thread_checker_.CalledOnValidThread());
   generic_event_map->swap(generic_event_map_);
   generic_event_map_.clear();
+  memset(next_write_index_, 0, sizeof(next_write_index_));
+}
+
+bool EncodingEventSubscriber::ShouldProcessEvent(CastLoggingEvent event) {
+  return GetEventMediaType(event) == config_.event_media_type;
+}
+
+void EncodingEventSubscriber::TruncateFrameEventMapIfNeeded() {
+  // This works because this is called everytime an event is inserted and
+  // we only insert events one at a time.
+  if (frame_event_map_.size() > config_.max_frames)
+    frame_event_map_.erase(frame_event_map_.begin());
+}
+
+void EncodingEventSubscriber::TruncatePacketEventMapIfNeeded() {
+  // This works because this is called everytime an event is inserted and
+  // we only insert events one at a time.
+  if (packet_event_map_.size() > config_.max_frames)
+    packet_event_map_.erase(packet_event_map_.begin());
 }
 
 }  // namespace cast
 }  // namespace media