src/media/cast/receiver/frame_receiver.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "media/cast/receiver/frame_receiver.h"
   6
   7 #include <algorithm>
   8
   9 #include "base/big_endian.h"
  10 #include "base/bind.h"
  11 #include "base/logging.h"
  12 #include "base/message_loop/message_loop.h"
  13 #include "media/cast/cast_environment.h"
  14
  15 namespace {
  16 const int kMinSchedulingDelayMs = 1;
  17 }  // namespace
  18
  19 namespace media {
  20 namespace cast {
  21
  22 FrameReceiver::FrameReceiver(
  23     const scoped_refptr<CastEnvironment>& cast_environment,
  24     const FrameReceiverConfig& config,
  25     EventMediaType event_media_type,
  26     transport::PacedPacketSender* const packet_sender)
  27     : cast_environment_(cast_environment),
  28       packet_parser_(config.incoming_ssrc, config.rtp_payload_type),
  29       stats_(cast_environment->Clock()),
  30       event_media_type_(event_media_type),
  31       event_subscriber_(kReceiverRtcpEventHistorySize, event_media_type),
  32       rtp_timebase_(config.frequency),
  33       target_playout_delay_(
  34           base::TimeDelta::FromMilliseconds(config.rtp_max_delay_ms)),
  35       expected_frame_duration_(
  36           base::TimeDelta::FromSeconds(1) / config.max_frame_rate),
  37       reports_are_scheduled_(false),
  38       framer_(cast_environment->Clock(),
  39               this,
  40               config.incoming_ssrc,
  41               true,
  42               config.rtp_max_delay_ms * config.max_frame_rate / 1000),
  43       rtcp_(cast_environment_,
  44             NULL,
  45             NULL,
  46             packet_sender,
  47             &stats_,
  48             config.rtcp_mode,
  49             base::TimeDelta::FromMilliseconds(config.rtcp_interval),
  50             config.feedback_ssrc,
  51             config.incoming_ssrc,
  52             config.rtcp_c_name,
  53             event_media_type),
  54       is_waiting_for_consecutive_frame_(false),
  55       lip_sync_drift_(ClockDriftSmoother::GetDefaultTimeConstant()),
  56       weak_factory_(this) {
  57   DCHECK_GT(config.rtp_max_delay_ms, 0);
  58   DCHECK_GT(config.max_frame_rate, 0);
  59   decryptor_.Initialize(config.aes_key, config.aes_iv_mask);
  60   rtcp_.SetTargetDelay(target_playout_delay_);
  61   cast_environment_->Logging()->AddRawEventSubscriber(&event_subscriber_);
  62   memset(frame_id_to_rtp_timestamp_, 0, sizeof(frame_id_to_rtp_timestamp_));
  63 }
  64
  65 FrameReceiver::~FrameReceiver() {
  66   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  67   cast_environment_->Logging()->RemoveRawEventSubscriber(&event_subscriber_);
  68 }
  69
  70 void FrameReceiver::RequestEncodedFrame(
  71     const ReceiveEncodedFrameCallback& callback) {
  72   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  73   frame_request_queue_.push_back(callback);
  74   EmitAvailableEncodedFrames();
  75 }
  76
  77 bool FrameReceiver::ProcessPacket(scoped_ptr<Packet> packet) {
  78   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  79
  80   if (Rtcp::IsRtcpPacket(&packet->front(), packet->size())) {
  81     rtcp_.IncomingRtcpPacket(&packet->front(), packet->size());
  82   } else {
  83     RtpCastHeader rtp_header;
  84     const uint8* payload_data;
  85     size_t payload_size;
  86     if (!packet_parser_.ParsePacket(&packet->front(),
  87                                     packet->size(),
  88                                     &rtp_header,
  89                                     &payload_data,
  90                                     &payload_size)) {
  91       return false;
  92     }
  93
  94     ProcessParsedPacket(rtp_header, payload_data, payload_size);
  95     stats_.UpdateStatistics(rtp_header);
  96   }
  97
  98   if (!reports_are_scheduled_) {
  99     ScheduleNextRtcpReport();
 100     ScheduleNextCastMessage();
 101     reports_are_scheduled_ = true;
 102   }
 103
 104   return true;
 105 }
 106
 107 // static
 108 bool FrameReceiver::ParseSenderSsrc(const uint8* packet,
 109                                     size_t length,
 110                                     uint32* ssrc) {
 111   base::BigEndianReader big_endian_reader(
 112       reinterpret_cast<const char*>(packet), length);
 113   return big_endian_reader.Skip(8) && big_endian_reader.ReadU32(ssrc);
 114 }
 115
 116 void FrameReceiver::ProcessParsedPacket(const RtpCastHeader& rtp_header,
 117                                         const uint8* payload_data,
 118                                         size_t payload_size) {
 119   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 120
 121   const base::TimeTicks now = cast_environment_->Clock()->NowTicks();
 122
 123   frame_id_to_rtp_timestamp_[rtp_header.frame_id & 0xff] =
 124       rtp_header.rtp_timestamp;
 125   cast_environment_->Logging()->InsertPacketEvent(
 126       now, PACKET_RECEIVED, event_media_type_, rtp_header.rtp_timestamp,
 127       rtp_header.frame_id, rtp_header.packet_id, rtp_header.max_packet_id,
 128       payload_size);
 129
 130   bool duplicate = false;
 131   const bool complete =
 132       framer_.InsertPacket(payload_data, payload_size, rtp_header, &duplicate);
 133
 134   // Duplicate packets are ignored.
 135   if (duplicate)
 136     return;
 137
 138   // Update lip-sync values upon receiving the first packet of each frame, or if
 139   // they have never been set yet.
 140   if (rtp_header.packet_id == 0 || lip_sync_reference_time_.is_null()) {
 141     RtpTimestamp fresh_sync_rtp;
 142     base::TimeTicks fresh_sync_reference;
 143     if (!rtcp_.GetLatestLipSyncTimes(&fresh_sync_rtp, &fresh_sync_reference)) {
 144       // HACK: The sender should have provided Sender Reports before the first
 145       // frame was sent.  However, the spec does not currently require this.
 146       // Therefore, when the data is missing, the local clock is used to
 147       // generate reference timestamps.
 148       VLOG(2) << "Lip sync info missing.  Falling-back to local clock.";
 149       fresh_sync_rtp = rtp_header.rtp_timestamp;
 150       fresh_sync_reference = now;
 151     }
 152     // |lip_sync_reference_time_| is always incremented according to the time
 153     // delta computed from the difference in RTP timestamps.  Then,
 154     // |lip_sync_drift_| accounts for clock drift and also smoothes-out any
 155     // sudden/discontinuous shifts in the series of reference time values.
 156     if (lip_sync_reference_time_.is_null()) {
 157       lip_sync_reference_time_ = fresh_sync_reference;
 158     } else {
 159       lip_sync_reference_time_ += RtpDeltaToTimeDelta(
 160           static_cast<int32>(fresh_sync_rtp - lip_sync_rtp_timestamp_),
 161           rtp_timebase_);
 162     }
 163     lip_sync_rtp_timestamp_ = fresh_sync_rtp;
 164     lip_sync_drift_.Update(
 165         now, fresh_sync_reference - lip_sync_reference_time_);
 166   }
 167
 168   // Another frame is complete from a non-duplicate packet.  Attempt to emit
 169   // more frames to satisfy enqueued requests.
 170   if (complete)
 171     EmitAvailableEncodedFrames();
 172 }
 173
 174 void FrameReceiver::CastFeedback(const RtcpCastMessage& cast_message) {
 175   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 176
 177   base::TimeTicks now = cast_environment_->Clock()->NowTicks();
 178   RtpTimestamp rtp_timestamp =
 179       frame_id_to_rtp_timestamp_[cast_message.ack_frame_id_ & 0xff];
 180   cast_environment_->Logging()->InsertFrameEvent(
 181       now, FRAME_ACK_SENT, event_media_type_,
 182       rtp_timestamp, cast_message.ack_frame_id_);
 183
 184   ReceiverRtcpEventSubscriber::RtcpEventMultiMap rtcp_events;
 185   event_subscriber_.GetRtcpEventsAndReset(&rtcp_events);
 186   rtcp_.SendRtcpFromRtpReceiver(&cast_message, &rtcp_events);
 187 }
 188
 189 void FrameReceiver::EmitAvailableEncodedFrames() {
 190   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 191
 192   while (!frame_request_queue_.empty()) {
 193     // Attempt to peek at the next completed frame from the |framer_|.
 194     // TODO(miu): We should only be peeking at the metadata, and not copying the
 195     // payload yet!  Or, at least, peek using a StringPiece instead of a copy.
 196     scoped_ptr<transport::EncodedFrame> encoded_frame(
 197         new transport::EncodedFrame());
 198     bool is_consecutively_next_frame = false;
 199     bool have_multiple_complete_frames = false;
 200     if (!framer_.GetEncodedFrame(encoded_frame.get(),
 201                                  &is_consecutively_next_frame,
 202                                  &have_multiple_complete_frames)) {
 203       VLOG(1) << "Wait for more packets to produce a completed frame.";
 204       return;  // ProcessParsedPacket() will invoke this method in the future.
 205     }
 206
 207     const base::TimeTicks now = cast_environment_->Clock()->NowTicks();
 208     const base::TimeTicks playout_time =
 209         GetPlayoutTime(encoded_frame->rtp_timestamp);
 210
 211     // If we have multiple decodable frames, and the current frame is
 212     // too old, then skip it and decode the next frame instead.
 213     if (have_multiple_complete_frames && now > playout_time) {
 214       framer_.ReleaseFrame(encoded_frame->frame_id);
 215       continue;
 216     }
 217
 218     // If |framer_| has a frame ready that is out of sequence, examine the
 219     // playout time to determine whether it's acceptable to continue, thereby
 220     // skipping one or more frames.  Skip if the missing frame wouldn't complete
 221     // playing before the start of playback of the available frame.
 222     if (!is_consecutively_next_frame) {
 223       // TODO(miu): Also account for expected decode time here?
 224       const base::TimeTicks earliest_possible_end_time_of_missing_frame =
 225           now + expected_frame_duration_;
 226       if (earliest_possible_end_time_of_missing_frame < playout_time) {
 227         VLOG(1) << "Wait for next consecutive frame instead of skipping.";
 228         if (!is_waiting_for_consecutive_frame_) {
 229           is_waiting_for_consecutive_frame_ = true;
 230           cast_environment_->PostDelayedTask(
 231               CastEnvironment::MAIN,
 232               FROM_HERE,
 233               base::Bind(&FrameReceiver::EmitAvailableEncodedFramesAfterWaiting,
 234                          weak_factory_.GetWeakPtr()),
 235               playout_time - now);
 236         }
 237         return;
 238       }
 239     }
 240
 241     // Decrypt the payload data in the frame, if crypto is being used.
 242     if (decryptor_.initialized()) {
 243       std::string decrypted_data;
 244       if (!decryptor_.Decrypt(encoded_frame->frame_id,
 245                               encoded_frame->data,
 246                               &decrypted_data)) {
 247         // Decryption failed.  Give up on this frame.
 248         framer_.ReleaseFrame(encoded_frame->frame_id);
 249         continue;
 250       }
 251       encoded_frame->data.swap(decrypted_data);
 252     }
 253
 254     // At this point, we have a decrypted EncodedFrame ready to be emitted.
 255     encoded_frame->reference_time = playout_time;
 256     framer_.ReleaseFrame(encoded_frame->frame_id);
 257     cast_environment_->PostTask(CastEnvironment::MAIN,
 258                                 FROM_HERE,
 259                                 base::Bind(frame_request_queue_.front(),
 260                                            base::Passed(&encoded_frame)));
 261     frame_request_queue_.pop_front();
 262   }
 263 }
 264
 265 void FrameReceiver::EmitAvailableEncodedFramesAfterWaiting() {
 266   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 267   DCHECK(is_waiting_for_consecutive_frame_);
 268   is_waiting_for_consecutive_frame_ = false;
 269   EmitAvailableEncodedFrames();
 270 }
 271
 272 base::TimeTicks FrameReceiver::GetPlayoutTime(uint32 rtp_timestamp) const {
 273   return lip_sync_reference_time_ +
 274       lip_sync_drift_.Current() +
 275       RtpDeltaToTimeDelta(
 276           static_cast<int32>(rtp_timestamp - lip_sync_rtp_timestamp_),
 277           rtp_timebase_) +
 278       target_playout_delay_;
 279 }
 280
 281 void FrameReceiver::ScheduleNextCastMessage() {
 282   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 283   base::TimeTicks send_time;
 284   framer_.TimeToSendNextCastMessage(&send_time);
 285   base::TimeDelta time_to_send =
 286       send_time - cast_environment_->Clock()->NowTicks();
 287   time_to_send = std::max(
 288       time_to_send, base::TimeDelta::FromMilliseconds(kMinSchedulingDelayMs));
 289   cast_environment_->PostDelayedTask(
 290       CastEnvironment::MAIN,
 291       FROM_HERE,
 292       base::Bind(&FrameReceiver::SendNextCastMessage,
 293                  weak_factory_.GetWeakPtr()),
 294       time_to_send);
 295 }
 296
 297 void FrameReceiver::SendNextCastMessage() {
 298   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 299   framer_.SendCastMessage();  // Will only send a message if it is time.
 300   ScheduleNextCastMessage();
 301 }
 302
 303 void FrameReceiver::ScheduleNextRtcpReport() {
 304   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 305   base::TimeDelta time_to_next = rtcp_.TimeToSendNextRtcpReport() -
 306                                  cast_environment_->Clock()->NowTicks();
 307
 308   time_to_next = std::max(
 309       time_to_next, base::TimeDelta::FromMilliseconds(kMinSchedulingDelayMs));
 310
 311   cast_environment_->PostDelayedTask(
 312       CastEnvironment::MAIN,
 313       FROM_HERE,
 314       base::Bind(&FrameReceiver::SendNextRtcpReport,
 315                  weak_factory_.GetWeakPtr()),
 316       time_to_next);
 317 }
 318
 319 void FrameReceiver::SendNextRtcpReport() {
 320   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 321   rtcp_.SendRtcpFromRtpReceiver(NULL, NULL);
 322   ScheduleNextRtcpReport();
 323 }
 324
 325 }  // namespace cast
 326 }  // namespace media