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