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