Cast: Implemented size limiting and event filtering in EncodingEventSubscriber.

media/cast/logging/encoding_event_subscriber.h

 // 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.
 
 #ifndef MEDIA_CAST_LOGGING_ENCODING_EVENT_SUBSCRIBER_H_
 #define MEDIA_CAST_LOGGING_ENCODING_EVENT_SUBSCRIBER_H_
 
 #include <map>
 
 #include "base/memory/linked_ptr.h"
 #include "base/threading/thread_checker.h"
+#include "media/cast/logging/logging_defines.h"
 #include "media/cast/logging/proto/raw_events.pb.h"
 #include "media/cast/logging/raw_event_subscriber.h"
 
 namespace media {
 namespace cast {
 
 typedef std::map<RtpTimestamp,
                  linked_ptr<media::cast::proto::AggregatedFrameEvent> >
     FrameEventMap;
 typedef std::map<RtpTimestamp,
                  linked_ptr<media::cast::proto::AggregatedPacketEvent> >
     PacketEventMap;
 typedef std::map<CastLoggingEvent,
                  linked_ptr<media::cast::proto::AggregatedGenericEvent> >
     GenericEventMap;
 
+// This struct is passed into EncodingEventSubscriber's constructor.
+struct EncodingEventSubscriberConfig {

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

Why not give the arguments directly to the constructor of
EncodingEventSubscriber?

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

I was thinking that with 3 parameters it's better to put them in a struct to
avoid cluttering the constructor. Since we might be getting rid of 1 of the
parameters and I'm not sure if we will be adding extra args in the future, I
will change it to pass the args directly to ctor.

+  // The subscriber will only process events that belong in |event_media_type|.

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

nit: that corresponds to |event_media_type|.

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

Done.

+  EventMediaType event_media_type;
+
+  // How many events to keep in the frame / packet map.
+  // This helps keep memory usage bounded.
+  // Every time one of |OnReceive[Frame,Packet]Event()| is
+  // called, it will check if the respective map size has exceeded |max_frames|.
+  // If so, it will remove the oldest aggregated entry (ordered by RTP
+  // timestamp).
+  // Note that there is no active process to remove old events periodically.

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

What does this mean?

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

Removing this. I was implementing windowing based on time earlier, so this
comment no longer applies. It referred to that there is no periodic timer to
remove events older than a certain time.

+  size_t max_frames;
+
+  // How many time-value pairs to keep per generic event.

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

I'm not sure we care about the generic events.

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

These are the generic events currently:

kRttMs, kPacketLoss, kJitterMs, kRembBitrate, kVideoAckReceived, kAudioAckSent,
kVideoAckSent.

You mentioned that we are not interested in the first 4. The last 3 seems like a
mistake to me - they should probably be frame events. I filed bug 339590 for
kAudioAckSent, kVideoAckSent. I am going to file a bug for kVideoAckReceived.

In the mean time, I will remove this param and stop listening to generic events.
Can add back later if needed.

+  // This helps keep memory usage bounded.
+  // Every time OnReceiveEvent() is called, it will check if the number of
+  // time-value pairs have exceeded the |max_generic_event_values|. If so,
+  // the oldest pair will be removed. Note that the time here refers to the
+  // event time as opposed to RTP timestamp.
+  // Note that there is no active process to remove old events periodically.
+  int max_generic_event_values;
+};
+
 // A RawEventSubscriber implementation that subscribes to events,
 // encodes them in protocol buffer format, and aggregates them into a more
 // compact structure.
-// TODO(imcheng): Implement event filtering and windowing based on size.
 class EncodingEventSubscriber : public RawEventSubscriber {
  public:
-  EncodingEventSubscriber();
+  EncodingEventSubscriber(EncodingEventSubscriberConfig config);
 
   virtual ~EncodingEventSubscriber();
 
   // RawReventSubscriber implementations.
   virtual void OnReceiveFrameEvent(const FrameEvent& frame_event) OVERRIDE;
   virtual void OnReceivePacketEvent(const PacketEvent& packet_event) OVERRIDE;
   virtual void OnReceiveGenericEvent(const GenericEvent& generic_event)
       OVERRIDE;
 
   // Assigns frame events received so far to |frame_events| and clears them
   // from this object.
   void GetFrameEventsAndReset(FrameEventMap* frame_events);
 
   // Assigns packet events received so far to |packet_events| and clears them
   // from this object.
   void GetPacketEventsAndReset(PacketEventMap* packet_events);
 
   // Assigns generic events received so far to |generic_events| and clears them
   // from this object.
   void GetGenericEventsAndReset(GenericEventMap* generic_events);
 
  private:
+  bool ShouldProcessEvent(CastLoggingEvent event);
+
+  // Removes oldest entry from |frame_event_map_| (ordered by RTP timestamp).
+  void TruncateFrameEventMapIfNeeded();
+
+  // Removes oldest entry from |packet_event_map_| (ordered by RTP timestamp).
+  void TruncatePacketEventMapIfNeeded();
+
+  const EncodingEventSubscriberConfig config_;
   FrameEventMap frame_event_map_;
   PacketEventMap packet_event_map_;
   GenericEventMap generic_event_map_;
 
+  // Used internally to keep track of next index to write the next time-value
+  // pair to for a particular generic event type.
+  int next_write_index_[kNumOfLoggingEvents];
+
   // All functions must be called on the main thread.
   base::ThreadChecker thread_checker_;
 
   DISALLOW_COPY_AND_ASSIGN(EncodingEventSubscriber);
 };
 
 }  // namespace cast
 }  // namespace media
 
 #endif  // MEDIA_CAST_LOGGING_ENCODING_EVENT_SUBSCRIBER_H_