Cast: an app to generate log data and do (de)serailization and (de)compression.

media/cast/test/log_encoder.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 <algorithm>
+#include <cstdio>
+#include <iostream>
+#include <string>
+#include <sstream>
+
+#include "base/rand_util.h"
+#include "base/time/default_tick_clock.h"
+#include "media/cast/cast_defines.h"
+#include "media/cast/logging/logging_defines.h"
+#include "media/cast/logging/logging_raw.h"
+#include "media/cast/logging/proto/raw_events.pb.h"
+#include "media/cast/test/fake_single_thread_task_runner.h"
+#include "media/cast/logging/encoding_event_subscriber.h"
+#include "net/base/big_endian.h"
+#include "third_party/zlib/zlib.h"
+
+using media::cast::CastLoggingEvent;
+using media::cast::EncodingEventSubscriber;
+using media::cast::FrameEvent;
+using media::cast::FrameEventMap;
+using media::cast::GenericEventMap;
+using media::cast::LoggingRaw;
+using media::cast::PacketEventMap;
+using media::cast::proto::AggregatedFrameEvent;
+using media::cast::proto::AggregatedGenericEvent;
+using media::cast::proto::AggregatedPacketEvent;
+using media::cast::proto::BasePacketEvent;
+using media::cast::test::FakeSingleThreadTaskRunner;
+
+const int kFps = 30;
+const int kNumSeconds = 60 * 30;
+
+const int kNumVideoFrames = kFps * kNumSeconds;
+const CastLoggingEvent kVideoFrameEvents[] = {
+    media::cast::kVideoFrameCaptured,
+    media::cast::kVideoFrameReceived,
+    media::cast::kVideoFrameSentToEncoder,
+    media::cast::kVideoFrameEncoded,
+    media::cast::kVideoFrameDecoded,
+    media::cast::kVideoRenderDelay};
+
+// Average number of packets per video frame.
+const int kNumPacketPerVideoFrame = 10;
+
+const CastLoggingEvent kVideoPacketEvents[] = {
+    media::cast::kPacketSentToPacer,
+    media::cast::kPacketSentToNetwork,
+    media::cast::kVideoPacketReceived};
+
+const CastLoggingEvent kOtherPacketEvents[] = {
+    media::cast::kPacketRetransmitted,
+    media::cast::kDuplicatePacketReceived};
+
+// 15 MB.
+const int kCompressedMaxSize = 10 * 1000 * 1000;
+// 100 MB.
+const int kDecompressedMaxSize = 100 * 1000 * 1000;
+
+// TODO(imcheng): All these should return bool indicating success.
+void WriteTo(const FrameEventMap& frame_events,
+             const PacketEventMap& packet_events,
+             uint8* out, int out_buf_size, int* out_size);
+void Compress(uint8* in, int in_buf_size, 
+              uint8* out, int out_buf_size, int* out_size);
+void Decompress(uint8* in, int in_buf_size, 
+                uint8* out, int out_buf_size, int* out_size);
+void ReadFrom(const uint8* in, const int in_size);
+
+int main(int argc, char** argv) {
+  EncodingEventSubscriber subscriber;
+  LoggingRaw logging_raw;
+
+  logging_raw.AddSubscriber(&subscriber);
+
+  base::DefaultTickClock clock;
+  base::TimeTicks base_now(clock.NowTicks());
+  base_now -= base::TimeDelta::FromSeconds(5400);
+  printf("Time now: %ld\n", base_now.ToInternalValue());
+
+  printf("\n===========\n");
+  printf("Inserting frame events\n");
+  base::TimeTicks now(base_now);
+  for (int i = 0; i < kNumVideoFrames; ++i) {
+    now += base::TimeDelta::FromMilliseconds(33);
+    base::TimeTicks base_now2(now);
+    uint32 rtp_timestamp = media::cast::GetVideoRtpTimestamp(now);
+    if (i < 5) {
+      printf("Inserting first 5 event with RTP timestamp %u\n", rtp_timestamp);
+    }
+    if (kNumVideoFrames - i < 5) {
+      printf("Inserting last 5 event with RTP timestamp %u\n", rtp_timestamp);
+    }
+    // Insert frame events for that frame.
+    for (uint32 event_index = 0; event_index < arraysize(kVideoFrameEvents);
+         ++event_index) {
+      now += base::TimeDelta::FromMilliseconds(5);
+      CastLoggingEvent type = kVideoFrameEvents[event_index];
+      if (type == media::cast::kVideoFrameEncoded) {
+        int size = base::RandInt(32768, 65536);
+        logging_raw.InsertFrameEventWithSize(now, type, rtp_timestamp, i, size);
+      } else if (type == media::cast::kVideoRenderDelay) {
+        base::TimeDelta delay(
+            base::TimeDelta::FromMilliseconds(base::RandInt(0, 100)));
+        logging_raw.InsertFrameEventWithDelay(now, type, rtp_timestamp, i,
+                                              delay);
+      } else {
+        logging_raw.InsertFrameEvent(now, type, rtp_timestamp, i);
+      }
+    }
+    now = base_now2;
+  }
+
+  printf("Inserting packet events\n");
+  now = base_now;
+  for (int i = 0; i < kNumVideoFrames; ++i) {
+    now += base::TimeDelta::FromMilliseconds(33);
+    base::TimeTicks base_now2(now);
+    uint32 rtp_timestamp = media::cast::GetVideoRtpTimestamp(now);
+
+    // Insert packet events for that frame.
+    for (int packet_id = 0; packet_id < kNumPacketPerVideoFrame; ++packet_id) {
+      int size = base::RandInt(2048, 4096);
+      for (uint32 event_index = 0; event_index < arraysize(kVideoPacketEvents);
+           ++event_index) {
+        now += base::TimeDelta::FromMilliseconds(1);
+        CastLoggingEvent type = kVideoPacketEvents[event_index];
+        logging_raw.InsertPacketEvent(now, type, rtp_timestamp, i, packet_id,
+                                      kNumPacketPerVideoFrame - 1, size);
+      }
+
+      // Randomly insert additional packet events.
+      int rand = base::RandInt(0, 99);
+      if (rand < 2) {
+        now += base::TimeDelta::FromMilliseconds(1);
+        logging_raw.InsertPacketEvent(now, media::cast::kPacketRetransmitted,
+                                      rtp_timestamp, i, packet_id,
+                                      kNumPacketPerVideoFrame - 1, size);
+      }
+      rand = base::RandInt(0, 99);
+      if (rand < 2) {
+        now += base::TimeDelta::FromMilliseconds(1);
+        logging_raw.InsertPacketEvent(
+            now, media::cast::kDuplicatePacketReceived, rtp_timestamp, i,
+            packet_id, kNumPacketPerVideoFrame - 1, size);
+      }
+    }
+
+    now = base_now2;
+  }
+
+  FrameEventMap frame_events;
+  subscriber.GetFrameEventsAndReset(&frame_events);
+
+  PacketEventMap packet_events;
+  subscriber.GetPacketEventsAndReset(&packet_events);
+
+  printf("Frame events map size: %lu\n", frame_events.size());
+  printf("Packet events map size: %lu\n", packet_events.size());
+
+  logging_raw.RemoveSubscriber(&subscriber);
+
+  printf("\n===========\n");
+  printf("Performing serialization\n");
+
+  scoped_ptr<uint8[]> serialized_buf(new uint8[kDecompressedMaxSize]);
+  int serialized_size;
+  WriteTo(frame_events, packet_events, serialized_buf.get(), 
+          kDecompressedMaxSize, &serialized_size);
+  printf("Serialized size: %d\n", serialized_size);
+
+  printf("\n===========\n");
+  printf("Performing compression\n");
+  scoped_ptr<uint8[]> compressed_buf(new uint8[kCompressedMaxSize]);
+  int compressed_size;
+  Compress(serialized_buf.get(), serialized_size, compressed_buf.get(),
+           kCompressedMaxSize, &compressed_size);
+  printf("Compressed size: %d\n", compressed_size);
+  serialized_buf.reset();
+
+  printf("\n===========\n");
+  printf("Performing decompression\n");
+  scoped_ptr<uint8[]> decompressed_buf(new uint8[kDecompressedMaxSize]);
+
+  int decompressed_size;
+  Decompress(compressed_buf.get(), compressed_size,
+             decompressed_buf.get(), kDecompressedMaxSize, &decompressed_size);
+  printf("Decompressed size: %d\n", decompressed_size);
+  compressed_buf.reset();
+
+  printf("\n===========\n");
+  printf("Performing deserialization\n");
+  ReadFrom(decompressed_buf.get(), decompressed_size);
+
+  printf("Done\n");
+  return 0;
+}
+
+void WriteTo(const FrameEventMap& frame_events,
+             const PacketEventMap& packet_events,
+             uint8* out, int out_buf_size, int* out_size) {
+
+  // int index = 0;
+  net::BigEndianWriter writer(out, out_buf_size);
+
+  // Frame events - write size first, then write entries
+  int frame_events_size = frame_events.size();
+  // memcpy(&out[index], &frame_events_size, sizeof(frame_events_size));
+  // index += sizeof(frame_events_size);
+  bool success;
+  success = writer.WriteU32(frame_events_size);
+  // XXX: instead of DCHECKs, just return false.
+  DCHECK(success);
+  int i = 0;
+  for (FrameEventMap::const_iterator it = frame_events.begin();
+       it != frame_events.end(); ++it) {
+    int proto_size = it->second->ByteSize();
+
+    if (frame_events_size - i < 5) {
+      printf("Serializing last 5 frame events, rtp ts: %u, %u\n", it->first, it->second->rtp_timestamp());
+    }
+    i++;
+    // Write size of the proto, then write the proto
+    // memcpy(&out[index], &proto_size, sizeof(proto_size));
+    // index += sizeof(proto_size);
+    success = writer.WriteU16(proto_size);
+    DCHECK(success);
+
+    // bool success = it->second->SerializeToArray(&out[index], out_size - index);
+    // DCHECK(success);
+    // index += proto_size;
+    success = it->second->SerializeToArray(writer.ptr(), writer.remaining());
+    DCHECK(success);
+    success = writer.Skip(proto_size);
+    DCHECK(success);
+  }
+
+  int frame_event_serialized_size = (char*)writer.ptr() - (char*)out;
+  printf("Encoded frame events size: %d\n", frame_event_serialized_size);
+
+  // Write packet events
+  int packet_event_size = packet_events.size();
+  // memcpy(&out[index], &packet_event_size, sizeof(packet_event_size));
+  // index += sizeof(packet_event_size);
+  success = writer.WriteU32(packet_event_size);
+  DCHECK(success);
+  i = 0;
+  for (PacketEventMap::const_iterator it = packet_events.begin();
+       it != packet_events.end(); ++it) {
+    int proto_size = it->second->ByteSize();
+    if (packet_event_size - i < 5) {
+      printf("Last 5 encoded packet size: %d, rtp timestamp: %u, %u\n", proto_size, it->first, it->second->rtp_timestamp());
+    }
+    i++;
+    // Write size of the proto, then write the proto
+    // memcpy(&out[index], &proto_size, sizeof(proto_size));
+    // index += sizeof(proto_size);
+    success = writer.WriteU16(proto_size);
+    DCHECK(success);
+
+    // bool success = it->second->SerializeToArray(&out[index], out_size - index);
+    // index += proto_size;
+    // DCHECK(success);
+    success = it->second->SerializeToArray(writer.ptr(), writer.remaining());
+    DCHECK(success);
+    success = writer.Skip(proto_size);
+    DCHECK(success);
+  }
+
+  int packet_event_serialized_size =
+      out_buf_size - writer.remaining() - frame_event_serialized_size;
+  printf("Encoded packet events buff count: %d\n",
+         packet_event_serialized_size);
+
+  *out_size = out_buf_size - writer.remaining();
+}
+
+void Compress(uint8* in, int in_buf_size, 
+              uint8* out, int out_buf_size, int* out_size) {
+  z_stream stream = {0};
+  int result = deflateInit2(&stream, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
+                            // 16 is added to produce a gzip header + trailer.
+                            MAX_WBITS + 16, 8,  // memLevel = 8 is default.
+                            Z_DEFAULT_STRATEGY);
+  DCHECK_EQ(Z_OK, result);
+  stream.next_in = in;
+  stream.avail_in = in_buf_size;
+  stream.next_out = out;
+  stream.avail_out = out_buf_size;
+  result = deflate(&stream, Z_FINISH);
+  DCHECK_EQ(Z_STREAM_END, result)
+      << ", maybe the output buffer isn't large enough?";
+  result = deflateEnd(&stream);
+  DCHECK_EQ(Z_OK, result);
+
+  *out_size = out_buf_size - stream.avail_out;
+}
+
+void Decompress(uint8* in, int in_buf_size, 
+                uint8* out, int out_buf_size, int* out_size) {
+  z_stream stream = {0};
+  // 16 is added to read in gzip format.
+  int result = inflateInit2(&stream, MAX_WBITS + 16);
+  DCHECK_EQ(Z_OK, result);
+  stream.next_in = in;
+  stream.avail_in = in_buf_size;
+  stream.next_out = out;
+  stream.avail_out = out_buf_size;
+  result = inflate(&stream, Z_FINISH);
+  DCHECK_EQ(Z_STREAM_END, result)
+      << ", maybe the output buffer isn't large enough?";
+  result = inflateEnd(&stream);
+  DCHECK_EQ(Z_OK, result);
+
+  // Trim to size.
+  *out_size = out_buf_size - stream.avail_out;
+}
+
+void ReadFrom(const uint8* in, const int in_size) {
+  //ArrayInputStream array_input_stream(&decompressed_string[0],
+  //                                    decompressed_string.size());
+  //CodedInputStream in(&array_input_stream);
+  
+// Reserve about 20MB.
+  const int out_size = 20 * 1000 * 1000;
+  scoped_ptr<uint8[]> out(new uint8[out_size]);
+  bool success;
+
+  //success = in.GetDirectBufferPointer(
+  //    reinterpret_cast<const void**>(&in_buffer), &buffer_size);
+  //DCHECK(success);
+
+  net::BigEndianReader reader(in, in_size);
+  uint32 num_frame_events;
+  //in_buffer = CodedInputStream::ReadLittleEndian32FromArray(in_buffer,
+  //                                                          &num_frame_events);
+  success = reader.ReadU32(&num_frame_events);
+  DCHECK(success);
+
+  printf("Read %u frame events\n", num_frame_events);
+
+  for (uint32 i = 0; i < num_frame_events; i++) {
+    uint16 proto_size = 0u;
+    //in_buffer =
+    //    CodedInputStream::ReadLittleEndian32FromArray(in_buffer, &proto_size);
+    success = reader.ReadU16(&proto_size);
+
+    AggregatedFrameEvent frame_event;
+    success = frame_event.ParseFromArray(reader.ptr(), proto_size);
+    DCHECK(success) << ", " << i;
+    reader.Skip(proto_size);
+    //in_buffer += proto_size;
+    if (i > num_frame_events - 5) {
+      printf("Last 5 event:\n");
+      printf("RTP timestamp: %u\n",
+             static_cast<uint32>(frame_event.rtp_timestamp()));
+      printf("Event type size: %d\n", frame_event.event_type_size());
+      printf("Event timestamp micros size: %d\n",
+             frame_event.event_timestamp_micros_size());
+      printf("Event encoded frame size: %d\n",
+             frame_event.encoded_frame_size());
+      printf("Event delay millis: %d\n", frame_event.delay_millis());
+    }
+  }
+
+  uint32 num_packet_events = 0u;
+  success = reader.ReadU32(&num_packet_events);
+  DCHECK(success);
+  //in_buffer = CodedInputStream::ReadLittleEndian32FromArray(in_buffer,
+  //                                                          &num_packet_events);
+  printf("Read %u packet events\n", num_packet_events);
+
+  for (uint32 i = 0; i < num_packet_events; i++) {
+    uint16 proto_size;
+    //in_buffer =
+    //    CodedInputStream::ReadLittleEndian32FromArray(in_buffer, &proto_size);
+    success = reader.ReadU16(&proto_size);
+    DCHECK(success);
+
+    AggregatedPacketEvent packet_event;
+    success = packet_event.ParseFromArray(reader.ptr(), proto_size);
+    DCHECK(success) << ", " << i;
+    reader.Skip(proto_size);
+
+    // in_buffer += proto_size;
+    if (i > num_packet_events - 5) {
+      printf("Last 5 event:\n");
+      printf("RTP timestamp: %u\n", packet_event.rtp_timestamp());
+      int base_size = packet_event.base_packet_event_size();
+      printf("Base packet event size: %d\n", base_size);
+      int base_packet_index = base::RandInt(0, base_size - 1);
+      printf("Getting base packet event %d\n", base_packet_index);
+      BasePacketEvent base_packet_event =
+          packet_event.base_packet_event(base_packet_index);
+      printf("Base Event %d timestamp packet id: %d\n", base_packet_index,
+             base_packet_event.packet_id());
+      printf("Base Event %d event type size: %d\n", base_packet_index,
+             base_packet_event.event_type_size());
+    }
+  }
+}