cast/streaming/receiver.cc

// Copyright 2019 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 "cast/streaming/receiver.h"

#include <algorithm>
#include <utility>

#include "absl/types/span.h"
#include "cast/streaming/constants.h"
#include "cast/streaming/receiver_packet_router.h"
#include "cast/streaming/session_config.h"
#include "util/chrono_helpers.h"
#include "util/osp_logging.h"
#include "util/std_util.h"

namespace openscreen {
namespace cast {

// Conveniences for ensuring logging output includes the SSRC of the Receiver,
// to help distinguish one out of multiple instances in a Cast Streaming
// session.
//
// TODO(miu): Replace RECEIVER_VLOG's with trace event logging once the tracing
// infrastructure is ready.
#define RECEIVER_LOG(level) OSP_LOG_##level << "[SSRC:" << ssrc() << "] "
#define RECEIVER_VLOG OSP_VLOG << "[SSRC:" << ssrc() << "] "

Receiver::Receiver(Environment* environment,
                   ReceiverPacketRouter* packet_router,
                   SessionConfig config)
    : now_(environment->now_function()),
      packet_router_(packet_router),
      config_(config),
      rtcp_session_(config.sender_ssrc, config.receiver_ssrc, now_()),
      rtcp_parser_(&rtcp_session_),
      rtcp_builder_(&rtcp_session_),
      stats_tracker_(config.rtp_timebase),
      rtp_parser_(config.sender_ssrc),
      rtp_timebase_(config.rtp_timebase),
      crypto_(config.aes_secret_key, config.aes_iv_mask),
      rtcp_buffer_capacity_(environment->GetMaxPacketSize()),
      rtcp_buffer_(new uint8_t[rtcp_buffer_capacity_]),
      rtcp_alarm_(environment->now_function(), environment->task_runner()),
      smoothed_clock_offset_(ClockDriftSmoother::kDefaultTimeConstant),
      consumption_alarm_(environment->now_function(),
                         environment->task_runner()) {
  OSP_DCHECK(packet_router_);
  OSP_DCHECK_EQ(checkpoint_frame(), FrameId::leader());
  OSP_CHECK_GT(rtcp_buffer_capacity_, 0);
  OSP_CHECK(rtcp_buffer_);

  rtcp_builder_.SetPlayoutDelay(config.target_playout_delay);
  playout_delay_changes_.emplace_back(FrameId::leader(),
                                      config.target_playout_delay);

  packet_router_->OnReceiverCreated(rtcp_session_.sender_ssrc(), this);
}

Receiver::~Receiver() {
  packet_router_->OnReceiverDestroyed(rtcp_session_.sender_ssrc());
}

void Receiver::SetConsumer(Consumer* consumer) {
  consumer_ = consumer;
  ScheduleFrameReadyCheck();
}

void Receiver::SetPlayerProcessingTime(Clock::duration needed_time) {
  player_processing_time_ = std::max(Clock::duration::zero(), needed_time);
}

void Receiver::RequestKeyFrame() {
  if (!last_key_frame_received_.is_null() &&
      last_frame_consumed_ >= last_key_frame_received_ &&
      !rtcp_builder_.is_picture_loss_indicator_set()) {
    rtcp_builder_.SetPictureLossIndicator(true);
    SendRtcp();
  }
}

int Receiver::AdvanceToNextFrame() {
  const FrameId immediate_next_frame = last_frame_consumed_ + 1;

  // Scan the queue for the next frame that should be consumed. Typically, this
  // is the very next frame; but if it is incomplete and already late for
  // playout, consider skipping-ahead.
  for (FrameId f = immediate_next_frame; f <= latest_frame_expected_; ++f) {
    PendingFrame& entry = GetQueueEntry(f);
    if (entry.collector.is_complete()) {
      const EncryptedFrame& encrypted_frame =
          entry.collector.PeekAtAssembledFrame();
      if (f == immediate_next_frame) {  // Typical case.
        RECEIVER_VLOG << "AdvanceToNextFrame: Next in sequence (" << f << ')';
        return FrameCrypto::GetPlaintextSize(encrypted_frame);
      }
      if (encrypted_frame.dependency != EncodedFrame::DEPENDS_ON_ANOTHER) {
        // Found a frame after skipping past some frames. Drop the ones being
        // skipped, advancing |last_frame_consumed_| before returning.
        RECEIVER_VLOG << "AdvanceToNextFrame: Skipping-ahead → " << f;
        DropAllFramesBefore(f);
        return FrameCrypto::GetPlaintextSize(encrypted_frame);
      }
      // Conclusion: The frame in the current queue entry is complete, but
      // depends on a prior incomplete frame. Continue scanning...
    }

    // Do not consider skipping past this frame if its estimated capture time is
    // unknown. The implication here is that, if |estimated_capture_time| is
    // set, the Receiver also knows whether any target playout delay changes
    // were communicated from the Sender in the frame's first RTP packet.
    if (!entry.estimated_capture_time) {
      break;
    }

    // If this incomplete frame is not yet late for playout, simply wait for the
    // rest of its packets to come in. However, do schedule a check to
    // re-examine things at the time it would become a late frame, to possibly
    // skip-over it.
    const auto playout_time =
        *entry.estimated_capture_time + ResolveTargetPlayoutDelay(f);
    if (playout_time > (now_() + player_processing_time_)) {
      ScheduleFrameReadyCheck(playout_time);
      break;
    }
  }

  RECEIVER_VLOG << "AdvanceToNextFrame: No frames ready. Last consumed was "
                << last_frame_consumed_ << '.';
  return kNoFramesReady;
}

EncodedFrame Receiver::ConsumeNextFrame(absl::Span<uint8_t> buffer) {
  // Assumption: The required call to AdvanceToNextFrame() ensures that
  // |last_frame_consumed_| is set to one before the frame to be consumed here.
  const FrameId frame_id = last_frame_consumed_ + 1;
  OSP_CHECK_LE(frame_id, checkpoint_frame());

  // Decrypt the frame, populating the given output |frame|.
  PendingFrame& entry = GetQueueEntry(frame_id);
  OSP_DCHECK(entry.collector.is_complete());
  EncodedFrame frame;
  frame.data = buffer;
  crypto_.Decrypt(entry.collector.PeekAtAssembledFrame(), &frame);
  OSP_DCHECK(entry.estimated_capture_time);
  frame.reference_time =
      *entry.estimated_capture_time + ResolveTargetPlayoutDelay(frame_id);

  RECEIVER_VLOG << "ConsumeNextFrame → " << frame.frame_id << ": "
                << frame.data.size() << " payload bytes, RTP Timestamp "
                << frame.rtp_timestamp
                       .ToTimeSinceOrigin<microseconds>(rtp_timebase_)
                       .count()
                << " µs, to play-out "
                << to_microseconds(frame.reference_time - now_()).count()
                << " µs from now.";

  entry.Reset();
  last_frame_consumed_ = frame_id;

  // Ensure the Consumer is notified if there are already more frames ready for
  // consumption, and it hasn't explicitly called AdvanceToNextFrame() to check
  // for itself.
  ScheduleFrameReadyCheck();

  return frame;
}

void Receiver::OnReceivedRtpPacket(Clock::time_point arrival_time,
                                   std::vector<uint8_t> packet) {
  const absl::optional<RtpPacketParser::ParseResult> part =
      rtp_parser_.Parse(packet);
  if (!part) {
    RECEIVER_LOG(WARN) << "Parsing of " << packet.size()
                       << " bytes as an RTP packet failed.";
    return;
  }
  stats_tracker_.OnReceivedValidRtpPacket(part->sequence_number,
                                          part->rtp_timestamp, arrival_time);

  // Ignore packets for frames the Receiver is no longer interested in.
  if (part->frame_id <= checkpoint_frame()) {
    return;
  }

  // Extend the range of frames known to this Receiver, within the capacity of
  // this Receiver's queue. Prepare the FrameCollectors to receive any
  // newly-discovered frames.
  if (part->frame_id > latest_frame_expected_) {
    const FrameId max_allowed_frame_id =
        last_frame_consumed_ + kMaxUnackedFrames;
    if (part->frame_id > max_allowed_frame_id) {
      RECEIVER_VLOG << "Dropping RTP packet for " << part->frame_id
                    << ": Too many frames are already in-flight.";
      return;
    }
    do {
      ++latest_frame_expected_;
      GetQueueEntry(latest_frame_expected_)
          .collector.set_frame_id(latest_frame_expected_);
    } while (latest_frame_expected_ < part->frame_id);
    RECEIVER_VLOG << "Advanced latest frame expected to "
                  << latest_frame_expected_;
  }

  // Start-up edge case: Blatantly drop the first packet of all frames until the
  // Receiver has processed at least one Sender Report containing the necessary
  // clock-drift and lip-sync information (see OnReceivedRtcpPacket()). This is
  // an inescapable data dependency. Note that this special case should almost
  // never trigger, since a well-behaving Sender will send the first Sender
  // Report RTCP packet before any of the RTP packets.
  if (!last_sender_report_ && part->packet_id == FramePacketId{0}) {
    RECEIVER_LOG(WARN) << "Dropping packet 0 of frame " << part->frame_id
                       << " because it arrived before the first Sender Report.";
    // Note: The Sender will have to re-transmit this dropped packet after the
    // Sender Report to allow the Receiver to move forward.
    return;
  }

  PendingFrame& pending_frame = GetQueueEntry(part->frame_id);
  FrameCollector& collector = pending_frame.collector;
  if (collector.is_complete()) {
    // An extra, redundant |packet| was received. Do nothing since the frame was
    // already complete.
    return;
  }

  if (!collector.CollectRtpPacket(*part, &packet)) {
    return;  // Bad data in the parsed packet. Ignore it.
  }

  // The first packet in a frame contains timing information critical for
  // computing this frame's (and all future frames') playout time. Process that,
  // but only once.
  if (part->packet_id == FramePacketId{0} &&
      !pending_frame.estimated_capture_time) {
    // Estimate the original capture time of this frame (at the Sender), in
    // terms of the Receiver's clock: First, start with a reference time point
    // from the Sender's clock (the one from the last Sender Report). Then,
    // translate it into the equivalent reference time point in terms of the
    // Receiver's clock by applying the measured offset between the two clocks.
    // Finally, apply the RTP timestamp difference between the Sender Report and
    // this frame to determine what the original capture time of this frame was.
    pending_frame.estimated_capture_time =
        last_sender_report_->reference_time + smoothed_clock_offset_.Current() +
        (part->rtp_timestamp - last_sender_report_->rtp_timestamp)
            .ToDuration<Clock::duration>(rtp_timebase_);

    // If a target playout delay change was included in this packet, record it.
    if (part->new_playout_delay > milliseconds::zero()) {
      RECEIVER_VLOG << "Target playout delay changes to "
                    << part->new_playout_delay.count() << " ms, as of "
                    << part->frame_id;
      RecordNewTargetPlayoutDelay(part->frame_id, part->new_playout_delay);
    }

    // Now that the estimated capture time is known, other frames may have just
    // become ready, per the frame-skipping logic in AdvanceToNextFrame().
    ScheduleFrameReadyCheck();
  }

  if (!collector.is_complete()) {
    return;  // Wait for the rest of the packets to come in.
  }
  const EncryptedFrame& encrypted_frame = collector.PeekAtAssembledFrame();

  // Whenever a key frame has been received, the decoder has what it needs to
  // recover. In this case, clear the PLI condition.
  if (encrypted_frame.dependency == EncryptedFrame::KEY_FRAME) {
    rtcp_builder_.SetPictureLossIndicator(false);
    last_key_frame_received_ = part->frame_id;
  }

  // If this just-completed frame is the one right after the checkpoint frame,
  // advance the checkpoint forward.
  if (part->frame_id == (checkpoint_frame() + 1)) {
    AdvanceCheckpoint(part->frame_id);
  }

  // Since a frame has become complete, schedule a check to see whether this or
  // any other frames have become ready for consumption.
  ScheduleFrameReadyCheck();
}

void Receiver::OnReceivedRtcpPacket(Clock::time_point arrival_time,
                                    std::vector<uint8_t> packet) {
  absl::optional<SenderReportParser::SenderReportWithId> parsed_report =
      rtcp_parser_.Parse(packet);
  if (!parsed_report) {
    RECEIVER_LOG(WARN) << "Parsing of " << packet.size()
                       << " bytes as an RTCP packet failed.";
    return;
  }
  last_sender_report_ = std::move(parsed_report);
  last_sender_report_arrival_time_ = arrival_time;

  // Measure the offset between the Sender's clock and the Receiver's Clock.
  // This will be used to translate reference timestamps from the Sender into
  // timestamps that represent the exact same moment in time at the Receiver.
  //
  // Note: Due to design limitations in the Cast Streaming spec, the Receiver
  // has no way to compute how long it took the Sender Report to travel over the
  // network. The calculation here just ignores that, and so the
  // |measured_offset| below will be larger than the true value by that amount.
  // This will have the effect of a later-than-configured playout delay.
  const Clock::duration measured_offset =
      arrival_time - last_sender_report_->reference_time;
  smoothed_clock_offset_.Update(arrival_time, measured_offset);
  RECEIVER_VLOG
      << "Received Sender Report: Local clock is ahead of Sender's by "
      << to_microseconds(smoothed_clock_offset_.Current()).count()
      << " µs (minus one-way network transit time).";

  RtcpReportBlock report;
  report.ssrc = rtcp_session_.sender_ssrc();
  stats_tracker_.PopulateNextReport(&report);
  report.last_status_report_id = last_sender_report_->report_id;
  report.SetDelaySinceLastReport(now_() - last_sender_report_arrival_time_);
  rtcp_builder_.IncludeReceiverReportInNextPacket(report);

  SendRtcp();
}

void Receiver::SendRtcp() {
  // Collect ACK/NACK feedback for all active frames in the queue.
  std::vector<PacketNack> packet_nacks;
  std::vector<FrameId> frame_acks;
  for (FrameId f = checkpoint_frame() + 1; f <= latest_frame_expected_; ++f) {
    const FrameCollector& collector = GetQueueEntry(f).collector;
    if (collector.is_complete()) {
      frame_acks.push_back(f);
    } else {
      collector.GetMissingPackets(&packet_nacks);
    }
  }

  // Build and send a compound RTCP packet.
  const bool no_nacks = packet_nacks.empty();
  rtcp_builder_.IncludeFeedbackInNextPacket(std::move(packet_nacks),
                                            std::move(frame_acks));
  last_rtcp_send_time_ = now_();
  packet_router_->SendRtcpPacket(rtcp_builder_.BuildPacket(
      last_rtcp_send_time_,
      absl::Span<uint8_t>(rtcp_buffer_.get(), rtcp_buffer_capacity_)));
  RECEIVER_VLOG << "Sent RTCP packet.";

  // Schedule the automatic sending of another RTCP packet, if this method is
  // not called within some bounded amount of time. While incomplete frames
  // exist in the queue, send RTCP packets (with ACK/NACK feedback) frequently.
  // When there are no incomplete frames, use a longer "keepalive" interval.
  const Clock::duration interval =
      (no_nacks ? kRtcpReportInterval : kNackFeedbackInterval);
  rtcp_alarm_.Schedule([this] { SendRtcp(); }, last_rtcp_send_time_ + interval);
}

const Receiver::PendingFrame& Receiver::GetQueueEntry(FrameId frame_id) const {
  return const_cast<Receiver*>(this)->GetQueueEntry(frame_id);
}

Receiver::PendingFrame& Receiver::GetQueueEntry(FrameId frame_id) {
  return pending_frames_[(frame_id - FrameId::first()) %
                         pending_frames_.size()];
}

void Receiver::RecordNewTargetPlayoutDelay(FrameId as_of_frame,
                                           milliseconds delay) {
  OSP_DCHECK_GT(as_of_frame, checkpoint_frame());

  // Prune-out entries from |playout_delay_changes_| that are no longer needed.
  // At least one entry must always be kept (i.e., there must always be a
  // "current" setting).
  const FrameId next_frame = last_frame_consumed_ + 1;
  const auto keep_one_before_it = std::find_if(
      std::next(playout_delay_changes_.begin()), playout_delay_changes_.end(),
      [&](const auto& entry) { return entry.first > next_frame; });
  playout_delay_changes_.erase(playout_delay_changes_.begin(),
                               std::prev(keep_one_before_it));

  // Insert the delay change entry, maintaining the ascending ordering of the
  // vector.
  const auto insert_it = std::find_if(
      playout_delay_changes_.begin(), playout_delay_changes_.end(),
      [&](const auto& entry) { return entry.first > as_of_frame; });
  playout_delay_changes_.emplace(insert_it, as_of_frame, delay);

  OSP_DCHECK(AreElementsSortedAndUnique(playout_delay_changes_));
}

milliseconds Receiver::ResolveTargetPlayoutDelay(FrameId frame_id) const {
  OSP_DCHECK_GT(frame_id, last_frame_consumed_);

#if OSP_DCHECK_IS_ON()
  // Extra precaution: Ensure all possible playout delay changes are known. In
  // other words, every unconsumed frame in the queue, up to (and including)
  // |frame_id|, must have an assigned estimated_capture_time.
  for (FrameId f = last_frame_consumed_ + 1; f <= frame_id; ++f) {
    OSP_DCHECK(GetQueueEntry(f).estimated_capture_time)
        << " don't know whether there was a playout delay change for frame "
        << f;
  }
#endif

  const auto it = std::find_if(
      playout_delay_changes_.crbegin(), playout_delay_changes_.crend(),
      [&](const auto& entry) { return entry.first <= frame_id; });
  OSP_DCHECK(it != playout_delay_changes_.crend());
  return it->second;
}

void Receiver::AdvanceCheckpoint(FrameId new_checkpoint) {
  OSP_DCHECK_GT(new_checkpoint, checkpoint_frame());
  OSP_DCHECK_LE(new_checkpoint, latest_frame_expected_);

  while (new_checkpoint < latest_frame_expected_) {
    const FrameId next = new_checkpoint + 1;
    if (!GetQueueEntry(next).collector.is_complete()) {
      break;
    }
    new_checkpoint = next;
  }

  RECEIVER_VLOG << "Advancing checkpoint to " << new_checkpoint;
  set_checkpoint_frame(new_checkpoint);
  rtcp_builder_.SetPlayoutDelay(ResolveTargetPlayoutDelay(new_checkpoint));
  SendRtcp();
}

void Receiver::DropAllFramesBefore(FrameId first_kept_frame) {
  // The following DCHECKs are verifying that this method is only being called
  // because one or more incomplete frames are being skipped-over.
  const FrameId first_to_drop = last_frame_consumed_ + 1;
  OSP_DCHECK_GT(first_kept_frame, first_to_drop);
  OSP_DCHECK_GT(first_kept_frame, checkpoint_frame());
  OSP_DCHECK_LE(first_kept_frame, latest_frame_expected_);

  // Reset each of the frames being dropped, pretending that they were consumed.
  for (FrameId f = first_to_drop; f < first_kept_frame; ++f) {
    PendingFrame& entry = GetQueueEntry(f);
    // Pedantic sanity-check: Ensure the "target playout delay change" data
    // dependency was satisfied. See comments in AdvanceToNextFrame().
    OSP_DCHECK(entry.estimated_capture_time);
    entry.Reset();
  }
  last_frame_consumed_ = first_kept_frame - 1;

  RECEIVER_LOG(INFO) << "Artificially advancing checkpoint after skipping.";
  AdvanceCheckpoint(first_kept_frame);
}

void Receiver::ScheduleFrameReadyCheck(Clock::time_point when) {
  consumption_alarm_.Schedule(
      [this] {
        if (consumer_) {
          const int next_frame_buffer_size = AdvanceToNextFrame();
          if (next_frame_buffer_size != kNoFramesReady) {
            consumer_->OnFramesReady(next_frame_buffer_size);
          }
        }
      },
      when);
}

Receiver::Consumer::~Consumer() = default;

Receiver::PendingFrame::PendingFrame() = default;
Receiver::PendingFrame::~PendingFrame() = default;

void Receiver::PendingFrame::Reset() {
  collector.Reset();
  estimated_capture_time = absl::nullopt;
}

// static
constexpr milliseconds Receiver::kDefaultPlayerProcessingTime;
constexpr int Receiver::kNoFramesReady;
constexpr milliseconds Receiver::kNackFeedbackInterval;

}  // namespace cast
}  // namespace openscreen