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     CastTransportSender* const transport)
  27     : cast_environment_(cast_environment),
  28       transport_(transport),
  29       packet_parser_(config.sender_ssrc, config.rtp_payload_type),
  30       stats_(cast_environment->Clock()),
  31       event_media_type_(event_media_type),
  32       event_subscriber_(kReceiverRtcpEventHistorySize, event_media_type),
  33       rtp_timebase_(config.rtp_timebase),
  34       target_playout_delay_(
  35           base::TimeDelta::FromMilliseconds(config.rtp_max_delay_ms)),
  36       expected_frame_duration_(
  37           base::TimeDelta::FromSeconds(1) / config.target_frame_rate),
  38       reports_are_scheduled_(false),
  39       framer_(cast_environment->Clock(),
  40               this,
  41               config.sender_ssrc,
  42               true,
  43               config.rtp_max_delay_ms * config.target_frame_rate / 1000),
  44       rtcp_(RtcpCastMessageCallback(),
  45             RtcpRttCallback(),
  46             RtcpLogMessageCallback(),
  47             cast_environment_->Clock(),
  48             NULL,
  49             config.receiver_ssrc,
  50             config.sender_ssrc),
  51       is_waiting_for_consecutive_frame_(false),
  52       lip_sync_drift_(ClockDriftSmoother::GetDefaultTimeConstant()),
  53       rtcp_interval_(base::TimeDelta::FromMilliseconds(config.rtcp_interval)),
  54       weak_factory_(this) {
  55   transport_->AddValidSsrc(config.sender_ssrc);
  56   DCHECK_GT(config.rtp_max_delay_ms, 0);
  57   DCHECK_GT(config.target_frame_rate, 0);
  58   decryptor_.Initialize(config.aes_key, config.aes_iv_mask);
  59   cast_environment_->Logging()->AddRawEventSubscriber(&event_subscriber_);
  60   memset(frame_id_to_rtp_timestamp_, 0, sizeof(frame_id_to_rtp_timestamp_));
  61 }
  62
  63 FrameReceiver::~FrameReceiver() {
  64   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  65   cast_environment_->Logging()->RemoveRawEventSubscriber(&event_subscriber_);
  66 }
  67
  68 void FrameReceiver::RequestEncodedFrame(
  69     const ReceiveEncodedFrameCallback& callback) {
  70   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  71   frame_request_queue_.push_back(callback);
  72   EmitAvailableEncodedFrames();
  73 }
  74
  75 bool FrameReceiver::ProcessPacket(scoped_ptr<Packet> packet) {
  76   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
  77
  78   if (Rtcp::IsRtcpPacket(&packet->front(), packet->size())) {
  79     rtcp_.IncomingRtcpPacket(&packet->front(), packet->size());
  80   } else {
  81     RtpCastHeader rtp_header;
  82     const uint8* payload_data;
  83     size_t payload_size;
  84     if (!packet_parser_.ParsePacket(&packet->front(),
  85                                     packet->size(),
  86                                     &rtp_header,
  87                                     &payload_data,
  88                                     &payload_size)) {
  89       return false;
  90     }
  91
  92     ProcessParsedPacket(rtp_header, payload_data, payload_size);
  93     stats_.UpdateStatistics(rtp_header);
  94   }
  95
  96   if (!reports_are_scheduled_) {
  97     ScheduleNextRtcpReport();
  98     ScheduleNextCastMessage();
  99     reports_are_scheduled_ = true;
 100   }
 101
 102   return true;
 103 }
 104
 105 void FrameReceiver::ProcessParsedPacket(const RtpCastHeader& rtp_header,
 106                                         const uint8* payload_data,
 107                                         size_t payload_size) {
 108   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 109
 110   const base::TimeTicks now = cast_environment_->Clock()->NowTicks();
 111
 112   frame_id_to_rtp_timestamp_[rtp_header.frame_id & 0xff] =
 113       rtp_header.rtp_timestamp;
 114   cast_environment_->Logging()->InsertPacketEvent(
 115       now, PACKET_RECEIVED, event_media_type_, rtp_header.rtp_timestamp,
 116       rtp_header.frame_id, rtp_header.packet_id, rtp_header.max_packet_id,
 117       payload_size);
 118
 119   bool duplicate = false;
 120   const bool complete =
 121       framer_.InsertPacket(payload_data, payload_size, rtp_header, &duplicate);
 122
 123   // Duplicate packets are ignored.
 124   if (duplicate)
 125     return;
 126
 127   // Update lip-sync values upon receiving the first packet of each frame, or if
 128   // they have never been set yet.
 129   if (rtp_header.packet_id == 0 || lip_sync_reference_time_.is_null()) {
 130     RtpTimestamp fresh_sync_rtp;
 131     base::TimeTicks fresh_sync_reference;
 132     if (!rtcp_.GetLatestLipSyncTimes(&fresh_sync_rtp, &fresh_sync_reference)) {
 133       // HACK: The sender should have provided Sender Reports before the first
 134       // frame was sent.  However, the spec does not currently require this.
 135       // Therefore, when the data is missing, the local clock is used to
 136       // generate reference timestamps.
 137       VLOG(2) << "Lip sync info missing.  Falling-back to local clock.";
 138       fresh_sync_rtp = rtp_header.rtp_timestamp;
 139       fresh_sync_reference = now;
 140     }
 141     // |lip_sync_reference_time_| is always incremented according to the time
 142     // delta computed from the difference in RTP timestamps.  Then,
 143     // |lip_sync_drift_| accounts for clock drift and also smoothes-out any
 144     // sudden/discontinuous shifts in the series of reference time values.
 145     if (lip_sync_reference_time_.is_null()) {
 146       lip_sync_reference_time_ = fresh_sync_reference;
 147     } else {
 148       lip_sync_reference_time_ += RtpDeltaToTimeDelta(
 149           static_cast<int32>(fresh_sync_rtp - lip_sync_rtp_timestamp_),
 150           rtp_timebase_);
 151     }
 152     lip_sync_rtp_timestamp_ = fresh_sync_rtp;
 153     lip_sync_drift_.Update(
 154         now, fresh_sync_reference - lip_sync_reference_time_);
 155   }
 156
 157   // Another frame is complete from a non-duplicate packet.  Attempt to emit
 158   // more frames to satisfy enqueued requests.
 159   if (complete)
 160     EmitAvailableEncodedFrames();
 161 }
 162
 163 void FrameReceiver::CastFeedback(const RtcpCastMessage& cast_message) {
 164   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 165
 166   base::TimeTicks now = cast_environment_->Clock()->NowTicks();
 167   RtpTimestamp rtp_timestamp =
 168       frame_id_to_rtp_timestamp_[cast_message.ack_frame_id & 0xff];
 169   cast_environment_->Logging()->InsertFrameEvent(
 170       now, FRAME_ACK_SENT, event_media_type_,
 171       rtp_timestamp, cast_message.ack_frame_id);
 172
 173   ReceiverRtcpEventSubscriber::RtcpEvents rtcp_events;
 174   event_subscriber_.GetRtcpEventsWithRedundancy(&rtcp_events);
 175   transport_->SendRtcpFromRtpReceiver(rtcp_.GetLocalSsrc(),
 176                                       rtcp_.GetRemoteSsrc(),
 177                                       rtcp_.ConvertToNTPAndSave(now),
 178                                       &cast_message,
 179                                       target_playout_delay_,
 180                                       &rtcp_events,
 181                                       NULL);
 182 }
 183
 184 void FrameReceiver::EmitAvailableEncodedFrames() {
 185   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 186
 187   while (!frame_request_queue_.empty()) {
 188     // Attempt to peek at the next completed frame from the |framer_|.
 189     // TODO(miu): We should only be peeking at the metadata, and not copying the
 190     // payload yet!  Or, at least, peek using a StringPiece instead of a copy.
 191     scoped_ptr<EncodedFrame> encoded_frame(
 192         new EncodedFrame());
 193     bool is_consecutively_next_frame = false;
 194     bool have_multiple_complete_frames = false;
 195     if (!framer_.GetEncodedFrame(encoded_frame.get(),
 196                                  &is_consecutively_next_frame,
 197                                  &have_multiple_complete_frames)) {
 198       VLOG(1) << "Wait for more packets to produce a completed frame.";
 199       return;  // ProcessParsedPacket() will invoke this method in the future.
 200     }
 201
 202     const base::TimeTicks now = cast_environment_->Clock()->NowTicks();
 203     const base::TimeTicks playout_time = GetPlayoutTime(*encoded_frame);
 204
 205     // If we have multiple decodable frames, and the current frame is
 206     // too old, then skip it and decode the next frame instead.
 207     if (have_multiple_complete_frames && now > playout_time) {
 208       framer_.ReleaseFrame(encoded_frame->frame_id);
 209       continue;
 210     }
 211
 212     // If |framer_| has a frame ready that is out of sequence, examine the
 213     // playout time to determine whether it's acceptable to continue, thereby
 214     // skipping one or more frames.  Skip if the missing frame wouldn't complete
 215     // playing before the start of playback of the available frame.
 216     if (!is_consecutively_next_frame) {
 217       // This assumes that decoding takes as long as playing, which might
 218       // not be true.
 219       const base::TimeTicks earliest_possible_end_time_of_missing_frame =
 220           now + expected_frame_duration_ * 2;
 221       if (earliest_possible_end_time_of_missing_frame < playout_time) {
 222         VLOG(1) << "Wait for next consecutive frame instead of skipping.";
 223         if (!is_waiting_for_consecutive_frame_) {
 224           is_waiting_for_consecutive_frame_ = true;
 225           cast_environment_->PostDelayedTask(
 226               CastEnvironment::MAIN,
 227               FROM_HERE,
 228               base::Bind(&FrameReceiver::EmitAvailableEncodedFramesAfterWaiting,
 229                          weak_factory_.GetWeakPtr()),
 230               playout_time - now);
 231         }
 232         return;
 233       }
 234     }
 235
 236     // At this point, we have the complete next frame, or a decodable
 237     // frame from somewhere later in the stream, AND we have given up
 238     // on waiting for any frames in between, so now we can ACK the frame.
 239     framer_.AckFrame(encoded_frame->frame_id);
 240
 241     // Decrypt the payload data in the frame, if crypto is being used.
 242     if (decryptor_.is_activated()) {
 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     if (encoded_frame->new_playout_delay_ms) {
 258       target_playout_delay_ = base::TimeDelta::FromMilliseconds(
 259           encoded_frame->new_playout_delay_ms);
 260     }
 261     cast_environment_->PostTask(CastEnvironment::MAIN,
 262                                 FROM_HERE,
 263                                 base::Bind(&FrameReceiver::EmitOneFrame,
 264                                            weak_factory_.GetWeakPtr(),
 265                                            frame_request_queue_.front(),
 266                                            base::Passed(&encoded_frame)));
 267     frame_request_queue_.pop_front();
 268   }
 269 }
 270
 271 void FrameReceiver::EmitAvailableEncodedFramesAfterWaiting() {
 272   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 273   DCHECK(is_waiting_for_consecutive_frame_);
 274   is_waiting_for_consecutive_frame_ = false;
 275   EmitAvailableEncodedFrames();
 276 }
 277
 278 void FrameReceiver::EmitOneFrame(const ReceiveEncodedFrameCallback& callback,
 279                                  scoped_ptr<EncodedFrame> encoded_frame) const {
 280   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 281   if (!callback.is_null())
 282     callback.Run(encoded_frame.Pass());
 283 }
 284
 285 base::TimeTicks FrameReceiver::GetPlayoutTime(const EncodedFrame& frame) const {
 286   base::TimeDelta target_playout_delay = target_playout_delay_;
 287   if (frame.new_playout_delay_ms) {
 288     target_playout_delay = base::TimeDelta::FromMilliseconds(
 289         frame.new_playout_delay_ms);
 290   }
 291   return lip_sync_reference_time_ +
 292       lip_sync_drift_.Current() +
 293       RtpDeltaToTimeDelta(
 294           static_cast<int32>(frame.rtp_timestamp - lip_sync_rtp_timestamp_),
 295           rtp_timebase_) +
 296       target_playout_delay;
 297 }
 298
 299 void FrameReceiver::ScheduleNextCastMessage() {
 300   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 301   base::TimeTicks send_time;
 302   framer_.TimeToSendNextCastMessage(&send_time);
 303   base::TimeDelta time_to_send =
 304       send_time - cast_environment_->Clock()->NowTicks();
 305   time_to_send = std::max(
 306       time_to_send, base::TimeDelta::FromMilliseconds(kMinSchedulingDelayMs));
 307   cast_environment_->PostDelayedTask(
 308       CastEnvironment::MAIN,
 309       FROM_HERE,
 310       base::Bind(&FrameReceiver::SendNextCastMessage,
 311                  weak_factory_.GetWeakPtr()),
 312       time_to_send);
 313 }
 314
 315 void FrameReceiver::SendNextCastMessage() {
 316   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 317   framer_.SendCastMessage();  // Will only send a message if it is time.
 318   ScheduleNextCastMessage();
 319 }
 320
 321 void FrameReceiver::ScheduleNextRtcpReport() {
 322   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 323   base::TimeDelta time_to_next = rtcp_interval_;
 324   time_to_next = std::max(
 325       time_to_next, base::TimeDelta::FromMilliseconds(kMinSchedulingDelayMs));
 326
 327   cast_environment_->PostDelayedTask(
 328       CastEnvironment::MAIN,
 329       FROM_HERE,
 330       base::Bind(&FrameReceiver::SendNextRtcpReport,
 331                  weak_factory_.GetWeakPtr()),
 332       time_to_next);
 333 }
 334
 335 void FrameReceiver::SendNextRtcpReport() {
 336   DCHECK(cast_environment_->CurrentlyOn(CastEnvironment::MAIN));
 337   const base::TimeTicks now = cast_environment_->Clock()->NowTicks();
 338   RtpReceiverStatistics stats = stats_.GetStatistics();
 339   transport_->SendRtcpFromRtpReceiver(rtcp_.GetLocalSsrc(),
 340                                       rtcp_.GetRemoteSsrc(),
 341                                       rtcp_.ConvertToNTPAndSave(now),
 342                                       NULL,
 343                                       base::TimeDelta(),
 344                                       NULL,
 345                                       &stats);
 346   ScheduleNextRtcpReport();
 347 }
 348
 349 }  // namespace cast
 350 }  // namespace media