Add callbacks for send channel rtp statistics

BUG=2235
R=mflodman@webrtc.org, pbos@webrtc.org, stefan@webrtc.org

Review URL: https://webrtc-codereview.appspot.com/4449004

git-svn-id: http://webrtc.googlecode.com/svn/trunk@5227 4adac7df-926f-26a2-2b94-8c16560cd09d

webrtc/common_types.h

@@ -235,7 +235,6 @@ protected:
 
 // Statistics for an RTCP channel
 struct RtcpStatistics {
- public:
   RtcpStatistics()
     : fraction_lost(0),
       cumulative_lost(0),
@@ -259,19 +258,20 @@ class RtcpStatisticsCallback {
 
 // Data usage statistics for a (rtp) stream
 struct StreamDataCounters {
- public:
   StreamDataCounters()
    : bytes(0),
+     header_bytes(0),
      padding_bytes(0),
      packets(0),
      retransmitted_packets(0),
      fec_packets(0) {}
 
-  uint32_t bytes;
+  uint32_t bytes;  // Payload bytes, excluding RTP headers and padding.
-  uint32_t padding_bytes;
+  uint32_t header_bytes;  // Number of bytes used by RTP headers.
-  uint32_t packets;
+  uint32_t padding_bytes;  // Number of padding bytes.
-  uint32_t retransmitted_packets;
+  uint32_t packets;  // Number of packets.
-  uint32_t fec_packets;
+  uint32_t retransmitted_packets;  // Number of retransmitted packets.
+  uint32_t fec_packets;  // Number of redundancy packets.
 };
 
 // Callback, called whenever byte/packet counts have been updated.
@@ -285,7 +285,6 @@ class StreamDataCountersCallback {
 
 // Rate statistics for a stream
 struct BitrateStatistics {
- public:
   BitrateStatistics()
     : bitrate_(0),
       packet_rate(0),

webrtc/modules/rtp_rtcp/interface/rtp_rtcp.h

@@ -338,6 +338,12 @@ class RtpRtcp : public Module {
     virtual bool GetSendSideDelay(int* avg_send_delay_ms,
                                   int* max_send_delay_ms) const = 0;
 
+    // Called on generation of new statistics after an RTP send.
+    virtual void RegisterSendChannelRtpStatisticsCallback(
+        StreamDataCountersCallback* callback) = 0;
+    virtual StreamDataCountersCallback*
+        GetSendChannelRtpStatisticsCallback() const = 0;
+
     /**************************************************************************
     *
     *   RTCP

webrtc/modules/rtp_rtcp/mocks/mock_rtp_rtcp.h

@@ -250,6 +250,10 @@ class MockRtpRtcp : public RtpRtcp {
       void(FrameCountObserver*));
   MOCK_CONST_METHOD0(GetSendFrameCountObserver,
       FrameCountObserver*(void));
+  MOCK_METHOD1(RegisterSendChannelRtpStatisticsCallback,
+      void(StreamDataCountersCallback*));
+  MOCK_CONST_METHOD0(GetSendChannelRtpStatisticsCallback,
+      StreamDataCountersCallback*(void));
   // Members.
   unsigned int remote_ssrc_;
 };

webrtc/modules/rtp_rtcp/source/rtp_rtcp_impl.cc

@@ -1667,6 +1667,16 @@ uint32_t ModuleRtpRtcpImpl::rtt_ms() const {
   return rtt_ms_;
 }
 
+void ModuleRtpRtcpImpl::RegisterSendChannelRtpStatisticsCallback(
+    StreamDataCountersCallback* callback) {
+  rtp_sender_.RegisterRtpStatisticsCallback(callback);
+}
+
+StreamDataCountersCallback*
+    ModuleRtpRtcpImpl::GetSendChannelRtpStatisticsCallback() const {
+  return rtp_sender_.GetRtpStatisticsCallback();
+}
+
 void ModuleRtpRtcpImpl::RegisterSendFrameCountObserver(
     FrameCountObserver* observer) {
   rtp_sender_.RegisterFrameCountObserver(observer);

webrtc/modules/rtp_rtcp/source/rtp_rtcp_impl.h

@@ -357,6 +357,11 @@ class ModuleRtpRtcpImpl : public RtpRtcp {
   virtual int32_t SendRTCPReferencePictureSelection(
       const uint64_t picture_id) OVERRIDE;
 
+  virtual void RegisterSendChannelRtpStatisticsCallback(
+      StreamDataCountersCallback* callback);
+  virtual StreamDataCountersCallback*
+      GetSendChannelRtpStatisticsCallback() const;
+
   void OnReceivedTMMBR();
 
   // Bad state of RTP receiver request a keyframe.

webrtc/modules/rtp_rtcp/source/rtp_sender.cc

@@ -55,13 +55,14 @@ RTPSender::RTPSender(const int32_t id, const bool audio, Clock *clock,
       packet_history_(clock),
       // Statistics
       statistics_crit_(CriticalSectionWrapper::CreateCriticalSection()),
-      packets_sent_(0), payload_bytes_sent_(0), start_time_stamp_forced_(false),
+      frame_count_observer_(NULL), rtp_stats_callback_(NULL),
+      // RTP variables
+      start_time_stamp_forced_(false),
       start_time_stamp_(0), ssrc_db_(*SSRCDatabase::GetSSRCDatabase()),
       remote_ssrc_(0), sequence_number_forced_(false), ssrc_forced_(false),
       timestamp_(0), capture_time_ms_(0), last_timestamp_time_ms_(0),
       last_packet_marker_bit_(false), num_csrcs_(0), csrcs_(),
-      include_csrcs_(true), rtx_(kRtxOff), payload_type_rtx_(-1),
+      include_csrcs_(true), rtx_(kRtxOff), payload_type_rtx_(-1) {
-      frame_counts_(), frame_count_observer_(NULL) {
   memset(nack_byte_count_times_, 0, sizeof(nack_byte_count_times_));
   memset(nack_byte_count_, 0, sizeof(nack_byte_count_));
   memset(csrcs_, 0, sizeof(csrcs_));
@@ -578,16 +579,6 @@ int32_t RTPSender::ReSendPacket(uint16_t packet_id, uint32_t min_resend_time) {
     return 0;
   }
 
-  {
-    // Update send statistics prior to pacer.
-    CriticalSectionScoped lock(statistics_crit_.get());
-    Bitrate::Update(length);
-    ++packets_sent_;
-    // We on purpose don't add to payload_bytes_sent_ since this is a
-    // re-transmit and not new payload data.
-  }
-
-
   ModuleRTPUtility::RTPHeaderParser rtp_parser(data_buffer, length);
   RTPHeader header;
   if (!rtp_parser.Parse(header)) {
@@ -620,6 +611,7 @@ int32_t RTPSender::ReSendPacket(uint16_t packet_id, uint32_t min_resend_time) {
   }
 
   if (SendPacketToNetwork(buffer_to_send_ptr, length)) {
+    UpdateRtpStats(buffer_to_send_ptr, length, header, rtx_ != kRtxOff, true);
     return length;
   }
   return -1;
@@ -813,7 +805,59 @@ bool RTPSender::PrepareAndSendPacket(uint8_t* buffer,
                                      rtp_header.sequenceNumber,
                                      rtp_header.headerLength);
   }
-  return SendPacketToNetwork(buffer_to_send_ptr, length);
+
+  bool ret = SendPacketToNetwork(buffer_to_send_ptr, length);
+  UpdateRtpStats(buffer_to_send_ptr, length, rtp_header, false, false);
+  return ret;
+}
+
+void RTPSender::UpdateRtpStats(const uint8_t* buffer,
+                               uint32_t size,
+                               const RTPHeader& header,
+                               bool is_rtx,
+                               bool is_retransmit) {
+  CriticalSectionScoped lock(statistics_crit_.get());
+  StreamDataCounters* counters;
+  uint32_t ssrc;
+  if (is_rtx) {
+    counters = &rtx_rtp_stats_;
+    ssrc = ssrc_rtx_;
+  } else {
+    counters = &rtp_stats_;
+    ssrc = ssrc_;
+  }
+
+  Bitrate::Update(size);
+  ++counters->packets;
+  if (IsFecPacket(buffer, header)) {
+    ++counters->fec_packets;
+  }
+
+  if (is_retransmit) {
+    ++counters->retransmitted_packets;
+  } else {
+    counters->bytes += size - (header.headerLength + header.paddingLength);
+    counters->header_bytes += header.headerLength;
+    counters->padding_bytes += header.paddingLength;
+  }
+
+  if (rtp_stats_callback_) {
+    rtp_stats_callback_->DataCountersUpdated(*counters, ssrc);
+  }
+}
+
+bool RTPSender::IsFecPacket(const uint8_t* buffer,
+                            const RTPHeader& header) const {
+  if (!video_) {
+    return false;
+  }
+  bool fec_enabled;
+  uint8_t pt_red;
+  uint8_t pt_fec;
+  video_->GenericFECStatus(fec_enabled, pt_red, pt_fec);
+  return fec_enabled &&
+      header.payloadType == pt_red &&
+      buffer[header.headerLength] == pt_fec;
 }
 
 int RTPSender::TimeToSendPadding(int bytes) {
@@ -875,13 +919,6 @@ int32_t RTPSender::SendToNetwork(
                                    storage) != 0) {
     return -1;
   }
-  {
-    // Update send statistics prior to pacer.
-    CriticalSectionScoped lock(statistics_crit_.get());
-    Bitrate::Update(payload_length + rtp_header_length);
-    ++packets_sent_;
-    payload_bytes_sent_ += payload_length;
-  }
 
   if (paced_sender_ && storage != kDontStore) {
     if (!paced_sender_->SendPacket(priority, rtp_header.ssrc,
@@ -895,10 +932,11 @@ int32_t RTPSender::SendToNetwork(
   if (capture_time_ms > 0) {
     UpdateDelayStatistics(capture_time_ms, now_ms);
   }
-  if (SendPacketToNetwork(buffer, payload_length + rtp_header_length)) {
+  uint32_t length = payload_length + rtp_header_length;
-    return 0;
+  if (!SendPacketToNetwork(buffer, length))
-  }
     return -1;
+  UpdateRtpStats(buffer, length, rtp_header, false, false);
+  return 0;
 }
 
 void RTPSender::UpdateDelayStatistics(int64_t capture_time_ms, int64_t now_ms) {
@@ -934,19 +972,23 @@ uint16_t RTPSender::IncrementSequenceNumber() {
 
 void RTPSender::ResetDataCounters() {
   CriticalSectionScoped lock(statistics_crit_.get());
-  packets_sent_ = 0;
+  rtp_stats_ = StreamDataCounters();
-  payload_bytes_sent_ = 0;
+  rtx_rtp_stats_ = StreamDataCounters();
+  if (rtp_stats_callback_) {
+    rtp_stats_callback_->DataCountersUpdated(rtp_stats_, ssrc_);
+    rtp_stats_callback_->DataCountersUpdated(rtx_rtp_stats_, ssrc_rtx_);
+  }
 }
 
 uint32_t RTPSender::Packets() const {
   CriticalSectionScoped lock(statistics_crit_.get());
-  return packets_sent_;
+  return rtp_stats_.packets + rtx_rtp_stats_.packets;
 }
 
 // Number of sent RTP bytes.
 uint32_t RTPSender::Bytes() const {
   CriticalSectionScoped lock(statistics_crit_.get());
-  return payload_bytes_sent_;
+  return rtp_stats_.bytes + rtx_rtp_stats_.bytes;
 }
 
 int RTPSender::CreateRTPHeader(
@@ -1533,4 +1575,17 @@ FrameCountObserver* RTPSender::GetFrameCountObserver() const {
   return frame_count_observer_;
 }
 
+void RTPSender::RegisterRtpStatisticsCallback(
+    StreamDataCountersCallback* callback) {
+  CriticalSectionScoped cs(statistics_crit_.get());
+  if (callback != NULL)
+    assert(rtp_stats_callback_ == NULL);
+  rtp_stats_callback_ = callback;
+}
+
+StreamDataCountersCallback* RTPSender::GetRtpStatisticsCallback() const {
+  CriticalSectionScoped cs(statistics_crit_.get());
+  return rtp_stats_callback_;
+}
+
 }  // namespace webrtc

webrtc/modules/rtp_rtcp/source/rtp_sender.h

@@ -271,6 +271,10 @@ class RTPSender : public Bitrate, public RTPSenderInterface {
                   int32_t bytes, StorageType store,
                   bool force_full_size_packets, bool only_pad_after_markerbit);
 
+  // Called on update of RTP statistics.
+  void RegisterRtpStatisticsCallback(StreamDataCountersCallback* callback);
+  StreamDataCountersCallback* GetRtpStatisticsCallback() const;
+
  protected:
   int32_t CheckPayloadType(const int8_t payload_type,
                            RtpVideoCodecTypes *video_type);
@@ -307,6 +311,13 @@ class RTPSender : public Bitrate, public RTPSenderInterface {
 
   void UpdateDelayStatistics(int64_t capture_time_ms, int64_t now_ms);
 
+  void UpdateRtpStats(const uint8_t* buffer,
+                      uint32_t size,
+                      const RTPHeader& header,
+                      bool is_rtx,
+                      bool is_retransmit);
+  bool IsFecPacket(const uint8_t* buffer, const RTPHeader& header) const;
+
   int32_t id_;
   const bool audio_configured_;
   RTPSenderAudio *audio_;
@@ -338,9 +349,12 @@ class RTPSender : public Bitrate, public RTPSenderInterface {
 
   // Statistics
   scoped_ptr<CriticalSectionWrapper> statistics_crit_;
-  uint32_t packets_sent_;
-  uint32_t payload_bytes_sent_;
   SendDelayMap send_delays_;
+  std::map<FrameType, uint32_t> frame_counts_;
+  FrameCountObserver* frame_count_observer_;
+  StreamDataCounters rtp_stats_;
+  StreamDataCounters rtx_rtp_stats_;
+  StreamDataCountersCallback* rtp_stats_callback_;
 
   // RTP variables
   bool start_time_stamp_forced_;
@@ -362,8 +376,6 @@ class RTPSender : public Bitrate, public RTPSenderInterface {
   int rtx_;
   uint32_t ssrc_rtx_;
   int payload_type_rtx_;
-  std::map<FrameType, uint32_t> frame_counts_;
-  FrameCountObserver* frame_count_observer_;
 };
 
 }  // namespace webrtc

webrtc/modules/rtp_rtcp/source/rtp_sender_unittest.cc

@@ -791,6 +791,85 @@ class RtpSenderAudioTest : public RtpSenderTest {
   }
 };
 
+TEST_F(RtpSenderTest, StreamDataCountersCallbacks) {
+  class TestCallback : public StreamDataCountersCallback {
+   public:
+    TestCallback()
+      : StreamDataCountersCallback(), ssrc_(0), counters_() {}
+    virtual ~TestCallback() {}
+
+    virtual void DataCountersUpdated(const StreamDataCounters& counters,
+                                     uint32_t ssrc) {
+      ssrc_ = ssrc;
+      counters_ = counters;
+    }
+
+    uint32_t ssrc_;
+    StreamDataCounters counters_;
+    bool Matches(uint32_t ssrc, uint32_t bytes, uint32_t header_bytes,
+                 uint32_t padding, uint32_t packets, uint32_t retransmits,
+                 uint32_t fec) {
+      return ssrc_ == ssrc &&
+          counters_.bytes == bytes &&
+          counters_.header_bytes == header_bytes &&
+          counters_.padding_bytes == padding &&
+          counters_.packets == packets &&
+          counters_.retransmitted_packets == retransmits &&
+          counters_.fec_packets == fec;
+    }
+
+  } callback;
+
+  const uint8_t kRedPayloadType = 96;
+  const uint8_t kUlpfecPayloadType = 97;
+  const uint32_t kMaxPaddingSize = 224;
+  char payload_name[RTP_PAYLOAD_NAME_SIZE] = "GENERIC";
+  const uint8_t payload_type = 127;
+  ASSERT_EQ(0, rtp_sender_->RegisterPayload(payload_name, payload_type, 90000,
+                                            0, 1500));
+  uint8_t payload[] = {47, 11, 32, 93, 89};
+  rtp_sender_->SetStorePacketsStatus(true, 1);
+  uint32_t ssrc = rtp_sender_->SSRC();
+
+  rtp_sender_->RegisterRtpStatisticsCallback(&callback);
+
+  // Send a frame.
+  ASSERT_EQ(0, rtp_sender_->SendOutgoingData(kVideoFrameKey, payload_type, 1234,
+                                             4321, payload, sizeof(payload),
+                                             NULL));
+
+  // {bytes = 6, header = 12, padding = 0, packets = 1, retrans = 0, fec = 0}
+  EXPECT_TRUE(callback.Matches(ssrc, 6, 12, 0, 1, 0, 0));
+
+  // Retransmit a frame.
+  uint16_t seqno = rtp_sender_->SequenceNumber() - 1;
+  rtp_sender_->ReSendPacket(seqno, 0);
+
+  // bytes = 6, header = 12, padding = 0, packets = 2, retrans = 1, fec = 0}
+  EXPECT_TRUE(callback.Matches(ssrc, 6, 12, 0, 2, 1, 0));
+
+  // Send padding.
+  rtp_sender_->TimeToSendPadding(kMaxPaddingSize);
+  // {bytes = 6, header = 24, padding = 224, packets = 3, retrans = 1, fec = 0}
+  EXPECT_TRUE(callback.Matches(ssrc, 6, 24, 224, 3, 1, 0));
+
+  // Send FEC.
+  rtp_sender_->SetGenericFECStatus(true, kRedPayloadType, kUlpfecPayloadType);
+  FecProtectionParams fec_params;
+  fec_params.fec_mask_type = kFecMaskRandom;
+  fec_params.fec_rate = 1;
+  fec_params.max_fec_frames = 1;
+  rtp_sender_->SetFecParameters(&fec_params, &fec_params);
+  ASSERT_EQ(0, rtp_sender_->SendOutgoingData(kVideoFrameDelta, payload_type,
+                                             1234, 4321, payload,
+                                             sizeof(payload), NULL));
+
+  // {bytes = 34, header = 48, padding = 224, packets = 5, retrans = 1, fec = 1}
+  EXPECT_TRUE(callback.Matches(ssrc, 34, 48, 224, 5, 1, 1));
+
+  rtp_sender_->RegisterRtpStatisticsCallback(NULL);
+}
+
 TEST_F(RtpSenderAudioTest, BuildRTPPacketWithAudioLevelExtension) {
   EXPECT_EQ(0, rtp_sender_->SetAudioLevelIndicationStatus(true,
       kAudioLevelExtensionId));

webrtc/video_engine/vie_channel.cc

@@ -358,6 +358,7 @@ int32_t ViEChannel::SetSendCodec(const VideoCodec& video_codec,
       rtp_rtcp->SetSendingMediaStatus(false);
       rtp_rtcp->RegisterSendFrameCountObserver(NULL);
       rtp_rtcp->RegisterSendChannelRtcpStatisticsCallback(NULL);
+      rtp_rtcp->RegisterSendChannelRtpStatisticsCallback(NULL);
       simulcast_rtp_rtcp_.pop_back();
       removed_rtp_rtcp_.push_front(rtp_rtcp);
     }
@@ -416,6 +417,8 @@ int32_t ViEChannel::SetSendCodec(const VideoCodec& video_codec,
           rtp_rtcp_->GetSendFrameCountObserver());
       rtp_rtcp->RegisterSendChannelRtcpStatisticsCallback(
           rtp_rtcp_->GetSendChannelRtcpStatisticsCallback());
+      rtp_rtcp->RegisterSendChannelRtpStatisticsCallback(
+          rtp_rtcp_->GetSendChannelRtpStatisticsCallback());
     }
     // |RegisterSimulcastRtpRtcpModules| resets all old weak pointers and old
     // modules can be deleted after this step.
@@ -428,6 +431,7 @@ int32_t ViEChannel::SetSendCodec(const VideoCodec& video_codec,
       rtp_rtcp->SetSendingMediaStatus(false);
       rtp_rtcp->RegisterSendFrameCountObserver(NULL);
       rtp_rtcp->RegisterSendChannelRtcpStatisticsCallback(NULL);
+      rtp_rtcp->RegisterSendChannelRtpStatisticsCallback(NULL);
       simulcast_rtp_rtcp_.pop_back();
       removed_rtp_rtcp_.push_front(rtp_rtcp);
     }
@@ -1349,6 +1353,21 @@ int32_t ViEChannel::GetRtpStatistics(uint32_t* bytes_sent,
   return 0;
 }
 
+void ViEChannel::RegisterSendChannelRtpStatisticsCallback(
+      StreamDataCountersCallback* callback) {
+  WEBRTC_TRACE(kTraceInfo, kTraceVideo, ViEId(engine_id_, channel_id_), "%s",
+                 __FUNCTION__);
+  rtp_rtcp_->RegisterSendChannelRtpStatisticsCallback(callback);
+  {
+    CriticalSectionScoped cs(rtp_rtcp_cs_.get());
+    for (std::list<RtpRtcp*>::iterator it = simulcast_rtp_rtcp_.begin();
+         it != simulcast_rtp_rtcp_.end();
+         it++) {
+      (*it)->RegisterSendChannelRtpStatisticsCallback(callback);
+    }
+  }
+}
+
 void ViEChannel::GetBandwidthUsage(uint32_t* total_bitrate_sent,
                                    uint32_t* video_bitrate_sent,
                                    uint32_t* fec_bitrate_sent,

webrtc/video_engine/vie_channel.h

@@ -190,6 +190,11 @@ class ViEChannel
                            uint32_t* packets_sent,
                            uint32_t* bytes_received,
                            uint32_t* packets_received) const;
+
+  // Called on update of RTP statistics.
+  void RegisterSendChannelRtpStatisticsCallback(
+      StreamDataCountersCallback* callback);
+
   void GetBandwidthUsage(uint32_t* total_bitrate_sent,
                          uint32_t* video_bitrate_sent,
                          uint32_t* fec_bitrate_sent,

webrtc/video_engine/vie_rtp_rtcp_impl.cc

@@ -1167,15 +1167,27 @@ int ViERTP_RTCPImpl::DeregisterReceiveChannelRtcpStatisticsCallback(
 }
 
 int ViERTP_RTCPImpl::RegisterSendChannelRtpStatisticsCallback(
-    int channel, StreamDataCountersCallback* callback) {
+    int video_channel, StreamDataCountersCallback* callback) {
-  // TODO(sprang): Implement
+  WEBRTC_TRACE(kTraceApiCall, kTraceVideo,
-  return -1;
+               ViEId(shared_data_->instance_id(), video_channel),
+               "%s(channel: %d)", __FUNCTION__, video_channel);
+  ViEChannelManagerScoped cs(*(shared_data_->channel_manager()));
+  ViEChannel* vie_channel = cs.Channel(video_channel);
+  assert(vie_channel != NULL);
+  vie_channel->RegisterSendChannelRtpStatisticsCallback(callback);
+  return 0;
 }
 
 int ViERTP_RTCPImpl::DeregisterSendChannelRtpStatisticsCallback(
-    int channel, StreamDataCountersCallback* callback) {
+    int video_channel, StreamDataCountersCallback* callback) {
-  // TODO(sprang): Implement
+  WEBRTC_TRACE(kTraceApiCall, kTraceVideo,
-  return -1;
+               ViEId(shared_data_->instance_id(), video_channel),
+               "%s(channel: %d)", __FUNCTION__, video_channel);
+  ViEChannelManagerScoped cs(*(shared_data_->channel_manager()));
+  ViEChannel* vie_channel = cs.Channel(video_channel);
+  assert(vie_channel != NULL);
+  vie_channel->RegisterSendChannelRtpStatisticsCallback(NULL);
+  return 0;
 }
 
 int ViERTP_RTCPImpl::RegisterReceiveChannelRtpStatisticsCallback(