Refactoring: CastTransport InitializeAudio/InitializeVideo.

media/cast/net/pacing/paced_sender.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/net/pacing/paced_sender.h"
 
 #include "base/big_endian.h"
 #include "base/bind.h"
 #include "base/debug/dump_without_crashing.h"
 #include "base/message_loop/message_loop.h"
@@ skipping 28 matching lines @@
                      uint16_t packet_id)
     : capture_time(capture_time),
       ssrc(ssrc),
       frame_id(frame_id),
       packet_id(packet_id) {}
 
 PacketKey::PacketKey(const PacketKey& other) = default;
 
 PacketKey::~PacketKey() {}
 
-PacedSender::PacketSendRecord::PacketSendRecord()
-    : last_byte_sent(0), last_byte_sent_for_audio(0), cancel_count(0) {}
+struct PacedSender::PacketSendRecord {
+  PacketSendRecord()
+      : last_byte_sent(0), last_byte_sent_for_audio(0), cancel_count(0) {}
+
+  base::TimeTicks time;    // Time when the packet was sent.
+  int64_t last_byte_sent;  // Number of bytes sent to network just after this
+                           // packet was sent.
+  int64_t last_byte_sent_for_audio;  // Number of bytes sent to network from
+                                     // audio stream just before this packet.
+  int cancel_count;  // Number of times the packet was canceled (debugging).
+};
+
+struct PacedSender::RtpSession {
+  explicit RtpSession(bool is_audio_stream)
+      : last_byte_sent(0), is_audio(is_audio_stream) {}
+  RtpSession() {}
+
+  // Tracks recently-logged RTP timestamps so that it can expand the truncated
+  // values found in packets.
+  RtpTimeTicks last_logged_rtp_timestamp_;
+  int64_t last_byte_sent;
+  bool is_audio;
+};
 
 PacedSender::PacedSender(
     size_t target_burst_size,
     size_t max_burst_size,
     base::TickClock* clock,
     std::vector<PacketEvent>* recent_packet_events,
     PacketTransport* transport,
     const scoped_refptr<base::SingleThreadTaskRunner>& transport_task_runner)
     : clock_(clock),
       recent_packet_events_(recent_packet_events),
       transport_(transport),
       transport_task_runner_(transport_task_runner),
-      audio_ssrc_(0),
-      video_ssrc_(0),
+      last_byte_sent_for_audio_(0),
       target_burst_size_(target_burst_size),
       max_burst_size_(max_burst_size),
       current_max_burst_size_(target_burst_size_),
       next_max_burst_size_(target_burst_size_),
       next_next_max_burst_size_(target_burst_size_),
       current_burst_size_(0),
       state_(State_Unblocked),
       has_reached_upper_bound_once_(false),
       weak_factory_(this) {}
 
 PacedSender::~PacedSender() {}
 
-void PacedSender::RegisterAudioSsrc(uint32_t audio_ssrc) {
-  audio_ssrc_ = audio_ssrc;
-}
+void PacedSender::RegisterSsrc(uint32_t ssrc, bool is_audio) {
+  if (sessions_.find(ssrc) != sessions_.end())
+    DVLOG(1) << "Re-register ssrc: " << ssrc;
 
-void PacedSender::RegisterVideoSsrc(uint32_t video_ssrc) {
-  video_ssrc_ = video_ssrc;
+  sessions_[ssrc] = RtpSession(is_audio);
 }
 
 void PacedSender::RegisterPrioritySsrc(uint32_t ssrc) {
   priority_ssrcs_.push_back(ssrc);
 }
 
 int64_t PacedSender::GetLastByteSentForPacket(const PacketKey& packet_key) {
   PacketSendHistory::const_iterator it = send_history_.find(packet_key);
   if (it == send_history_.end())
     return 0;
   return it->second.last_byte_sent;
 }
 
 int64_t PacedSender::GetLastByteSentForSsrc(uint32_t ssrc) {
-  std::map<uint32_t, int64_t>::const_iterator it = last_byte_sent_.find(ssrc);
-  if (it == last_byte_sent_.end())
+  auto it = sessions_.find(ssrc);
+  if (it == sessions_.end())
     return 0;
-  return it->second;
+  return it->second.last_byte_sent;
 }
 
 bool PacedSender::SendPackets(const SendPacketVector& packets) {
   if (packets.empty()) {
     return true;
   }
   const bool high_priority = IsHighPriority(packets.begin()->first);
   for (size_t i = 0; i < packets.size(); i++) {
     if (VLOG_IS_ON(2)) {
       PacketSendHistory::const_iterator history_it =
@@ skipping 32 matching lines @@
   if (it == send_history_.end())
     return true;
 
   // Suppose there is request to retransmit X and there is an audio
   // packet Y sent just before X. Reject retransmission of X if ACK for
   // Y has not been received.
   // Only do this for video packets.
   //
   // TODO(miu): This sounds wrong.  Audio packets are always transmitted first
   // (because they are put in |priority_packet_list_|, see PopNextPacket()).
-  if (packet_key.ssrc == video_ssrc_) {
+  auto session_it = sessions_.find(packet_key.ssrc);
+  DCHECK(session_it != sessions_.end());

[dcheng, 2016/07/14 03:40:27]

The inconsistent handling of ssrc here makes me nervous. Some things early
return and some DCHECK. It looks like this case is /probably/ OK, since
CastTransportImpl consults its own sessions_ map first... but I'm not really
sure. Are PacedSender::sessions_ and CastTransportImpl::sessions_ guaranteed to
be in sync?

[xjz, 2016/07/14 18:29:57]

Yes, whenever a session for a RTP stream (with unique ssrc) is inserted in
CastTransportImpl::sessions_, it will be registered in PacedSender::sessions_ as
well. A session should be always registered in both sessions_ through
CastTransportImpl::InitializeStream() before any transmission starts. 

[dcheng, 2016/07/15 01:21:19]

Is it possible to be consistent about DCHECKing() through this file? Or are some
of these function not guarded by a sessions_ check in CastTransportImpl?

[xjz, 2016/07/18 20:21:58]

Done. Removed the DCHECK() since it should be guarded by the check in
CastTransportImpl.

[dcheng, 2016/07/19 05:56:10]

To be clear, I think it's better to have the DCHECK: it's a useful assert to
have to trigger if this precondition is ever violated (and it's useful for
documenting said precondition). I was just trying to better understand how this
is all handled, since some sessions_.find() instances are guarded with an if ()
check and some are not.

For example, in LogPacketEvent, we have this check. But it's not clear why. For
example, I think this function can get called indirectly via SendFrameAsPackets.
But the ssrc in there appears to be configured in
CastTransportImpl::Initialize... so how can it be out of sync? Maybe it can, and
I don't understand how, but this is an example of what I mean.

[xjz, 2016/07/19 18:18:33]

Thanks for the clarification. Yes, the sessions should be already registered
through CastTransportImpl::InitializeStream(). Either DCHECK() or if() check is
only a guard for any possible future violation. I now use NOTREACHED() to
indicate that this should not happen. 

[xjz, 2016/07/19 21:43:19]

Changed it back to DCHECK() so that the code will crash if the condition is
violated (which should never happen). 

+  if (!session_it->second.is_audio) {
     if (dedup_info.last_byte_acked_for_audio &&
         it->second.last_byte_sent_for_audio &&
         dedup_info.last_byte_acked_for_audio <
         it->second.last_byte_sent_for_audio) {
       return false;
     }
   }
   // Retransmission interval has to be greater than |resend_interval|.
   if (now - it->second.time < dedup_info.resend_interval)
     return false;
@@ skipping 218 matching lines @@
         LogPacketEvent(packet->data, PACKET_SENT_TO_NETWORK);
         break;
       case PacketType_RTCP:
         break;
     }
 
     const bool socket_blocked = !transport_->SendPacket(packet, cb);
 
     // Save the send record.
     send_record->last_byte_sent = transport_->GetBytesSent();
-    send_record->last_byte_sent_for_audio = GetLastByteSentForSsrc(audio_ssrc_);
+    send_record->last_byte_sent_for_audio = last_byte_sent_for_audio_;
     send_history_buffer_[packet_key] = *send_record;
-    last_byte_sent_[packet_key.ssrc] = send_record->last_byte_sent;
+
+    auto it = sessions_.find(packet_key.ssrc);
+    DCHECK(it != sessions_.end());
+    it->second.last_byte_sent = send_record->last_byte_sent;
+    if (it->second.is_audio)
+      last_byte_sent_for_audio_ = send_record->last_byte_sent;
 
     if (socket_blocked) {
       state_ = State_TransportBlocked;
       return;
     }
     current_burst_size_++;
   }
 
   // Keep ~0.5 seconds of data (1000 packets).
   //
@@ skipping 21 matching lines @@
   // TODO(miu): This parsing logic belongs in RtpParser.
   event.timestamp = clock_->NowTicks();
   event.type = type;
   base::BigEndianReader reader(reinterpret_cast<const char*>(&packet[0]),
                                packet.size());
   bool success = reader.Skip(4);
   uint32_t truncated_rtp_timestamp;
   success &= reader.ReadU32(&truncated_rtp_timestamp);
   uint32_t ssrc;
   success &= reader.ReadU32(&ssrc);
-  if (ssrc == audio_ssrc_) {
-    event.rtp_timestamp = last_logged_audio_rtp_timestamp_ =
-        last_logged_audio_rtp_timestamp_.Expand(truncated_rtp_timestamp);
-    event.media_type = AUDIO_EVENT;
-  } else if (ssrc == video_ssrc_) {
-    event.rtp_timestamp = last_logged_video_rtp_timestamp_ =
-        last_logged_video_rtp_timestamp_.Expand(truncated_rtp_timestamp);
-    event.media_type = VIDEO_EVENT;
-  } else {
+
+  auto it = sessions_.find(ssrc);
+  if (it == sessions_.end()) {
     DVLOG(3) << "Got unknown ssrc " << ssrc << " when logging packet event";
     return;
   }
+  event.rtp_timestamp = it->second.last_logged_rtp_timestamp_ =
+      it->second.last_logged_rtp_timestamp_.Expand(truncated_rtp_timestamp);
+  event.media_type = it->second.is_audio ? AUDIO_EVENT : VIDEO_EVENT;
   success &= reader.Skip(2);
   success &= reader.ReadU16(&event.packet_id);
   success &= reader.ReadU16(&event.max_packet_id);
   event.size = base::checked_cast<uint32_t>(packet.size());
   DCHECK(success);
 }
 
 }  // namespace cast
 }  // namespace media