Media Remoting: Data/Control plumbing between renderer and Media Router.

chrome/browser/media/cast_remoting_sender.cc

 // Copyright 2016 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 "chrome/browser/media/cast_remoting_sender.h"
 
+#include <algorithm>
 #include <map>
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/callback.h"
 #include "base/lazy_instance.h"
 #include "base/memory/ptr_util.h"
 #include "base/numerics/safe_conversions.h"
+#include "base/stl_util.h"
 #include "base/trace_event/trace_event.h"
 #include "content/public/browser/browser_thread.h"
 #include "media/base/bind_to_current_loop.h"
 #include "media/cast/constants.h"
 
+using content::BrowserThread;
+
 namespace {
 
 // Global map for looking-up CastRemotingSender instances by their
-// |remoting_stream_id_|.
+// |rtp_stream_id|.
 using CastRemotingSenderMap = std::map<int32_t, cast::CastRemotingSender*>;
 base::LazyInstance<CastRemotingSenderMap>::Leaky g_sender_map =
     LAZY_INSTANCE_INITIALIZER;
 
 constexpr base::TimeDelta kMinSchedulingDelay =
     base::TimeDelta::FromMilliseconds(1);
 constexpr base::TimeDelta kMaxAckDelay = base::TimeDelta::FromMilliseconds(800);
-constexpr base::TimeDelta kMinAckDelay = base::TimeDelta::FromMilliseconds(400);
 constexpr base::TimeDelta kReceiverProcessTime =
     base::TimeDelta::FromMilliseconds(250);
 
 }  // namespace
 
 namespace cast {
 
 class CastRemotingSender::RemotingRtcpClient final
     : public media::cast::RtcpObserver {
  public:
@@ skipping 23 matching lines @@
   DISALLOW_COPY_AND_ASSIGN(RemotingRtcpClient);
 };
 
 // Convenience macro used in logging statements throughout this file.
 #define SENDER_SSRC (is_audio_ ? "AUDIO[" : "VIDEO[") << ssrc_ << "] "
 
 CastRemotingSender::CastRemotingSender(
     scoped_refptr<media::cast::CastEnvironment> cast_environment,
     media::cast::CastTransport* transport,
     const media::cast::CastTransportRtpConfig& config)
-    : remoting_stream_id_(config.rtp_stream_id),
+    : rtp_stream_id_(config.rtp_stream_id),
       cast_environment_(std::move(cast_environment)),
       transport_(transport),
       ssrc_(config.ssrc),
       is_audio_(config.rtp_payload_type ==
                 media::cast::RtpPayloadType::REMOTE_AUDIO),
+      binding_(this),
       max_ack_delay_(kMaxAckDelay),
       last_sent_frame_id_(media::cast::FrameId::first() - 1),
       latest_acked_frame_id_(media::cast::FrameId::first() - 1),
       duplicate_ack_counter_(0),
-      last_frame_was_canceled_(false),
+      flow_control_state_(RESTARTING),
       weak_factory_(this) {
+  // Confirm this constructor is running on the IO BrowserThread and the
+  // CastEnvironment::MAIN thread is the same thread.
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
   DCHECK(cast_environment_->CurrentlyOn(media::cast::CastEnvironment::MAIN));
 
-  CastRemotingSender*& pointer_in_map = g_sender_map.Get()[remoting_stream_id_];
+  CastRemotingSender*& pointer_in_map = g_sender_map.Get()[rtp_stream_id_];
   DCHECK(!pointer_in_map);
   pointer_in_map = this;
+
   transport_->InitializeStream(
       config, base::MakeUnique<RemotingRtcpClient>(weak_factory_.GetWeakPtr()));
 }
 
 CastRemotingSender::~CastRemotingSender() {
-  DCHECK(cast_environment_->CurrentlyOn(media::cast::CastEnvironment::MAIN));
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  g_sender_map.Pointer()->erase(rtp_stream_id_);
+}
 
-  g_sender_map.Pointer()->erase(remoting_stream_id_);
+// static
+void CastRemotingSender::FindAndBind(
+    int32_t rtp_stream_id,
+    mojo::ScopedDataPipeConsumerHandle pipe,
+    media::mojom::RemotingDataStreamSenderRequest request,
+    const base::Closure& error_callback) {
+  // CastRemotingSender lives entirely on the IO thread, so trampoline if
+  // necessary.
+  if (!BrowserThread::CurrentlyOn(BrowserThread::IO)) {
+    BrowserThread::PostTask(
+        BrowserThread::IO, FROM_HERE,
+        base::Bind(&CastRemotingSender::FindAndBind, rtp_stream_id,
+                   base::Passed(&pipe), base::Passed(&request),
+                   // Using media::BindToCurrentLoop() so the |error_callback|
+                   // is trampolined back to the original thread.
+                   media::BindToCurrentLoop(error_callback)));
+    return;
+  }
+
+  DCHECK(!error_callback.is_null());
+
+  // Look-up the CastRemotingSender instance by its |rtp_stream_id|.
+  const auto it = g_sender_map.Pointer()->find(rtp_stream_id);
+  if (it == g_sender_map.Pointer()->end()) {
+    DLOG(ERROR) << "Cannot find CastRemotingSender instance by ID: "
+                << rtp_stream_id;
+    error_callback.Run();
+    return;
+  }
+  CastRemotingSender* const sender = it->second;
+
+  // Confirm that the CastRemotingSender isn't already bound to a message pipe.
+  if (sender->binding_.is_bound()) {
+    DLOG(ERROR) << "Attempt to bind to CastRemotingSender a second time (id="
+                << rtp_stream_id << ")!";
+    error_callback.Run();
+    return;
+  }
+
+  DCHECK(sender->error_callback_.is_null());
+  sender->error_callback_ = error_callback;
+
+  sender->pipe_ = std::move(pipe);
+  sender->binding_.Bind(std::move(request));
+  sender->binding_.set_connection_error_handler(
+      base::Bind(&base::Closure::Run,
+                 base::Unretained(&sender->error_callback_)));
 }
 
 void CastRemotingSender::OnReceivedRtt(base::TimeDelta round_trip_time) {
   DCHECK_GT(round_trip_time, base::TimeDelta());
   current_round_trip_time_ = round_trip_time;
-  max_ack_delay_ = 2 * current_round_trip_time_ + kReceiverProcessTime;
-  max_ack_delay_ =
-      std::max(std::min(max_ack_delay_, kMaxAckDelay), kMinAckDelay);
+  max_ack_delay_ = 2 * std::max(current_round_trip_time_, base::TimeDelta()) +
+      kReceiverProcessTime;
+  max_ack_delay_ = std::min(max_ack_delay_, kMaxAckDelay);
 }
 
 void CastRemotingSender::ResendCheck() {
   DCHECK(cast_environment_->CurrentlyOn(media::cast::CastEnvironment::MAIN));
 
   DCHECK(!last_send_time_.is_null());
   const base::TimeDelta time_since_last_send =
       cast_environment_->Clock()->NowTicks() - last_send_time_;
   if (time_since_last_send > max_ack_delay_) {
     if (latest_acked_frame_id_ == last_sent_frame_id_) {
@@ skipping 118 matching lines @@
       frames_to_cancel.push_back(latest_acked_frame_id_);
       // This is a good place to match the trace for frame ids
       // since this ensures we not only track frame ids that are
       // implicitly ACKed, but also handles duplicate ACKs
       TRACE_EVENT_ASYNC_END1(
           "cast.stream", is_audio_ ? "Audio Transport" : "Video Transport",
           latest_acked_frame_id_.lower_32_bits(), "RTT_usecs",
           current_round_trip_time_.InMicroseconds());
     } while (latest_acked_frame_id_ < cast_feedback.ack_frame_id);
     transport_->CancelSendingFrames(ssrc_, frames_to_cancel);
+
+    if (flow_control_state_ == CONSUME_PAUSED) {
+      // One or more frames were canceled, so resume reading from the Mojo data
+      // pipes and sending frames.
+      binding_.ResumeIncomingMethodCallProcessing();
+      flow_control_state_ = CONSUMING;
+      VLOG(1) << SENDER_SSRC
+              << "Now CONSUMING because one or more frames were ACK'ed.";
+
+      // The last call to SendFrame() placed the sender into the CONSUME_PAUSED
+      // state without sending the frame. So, send that frame now.
+      SendFrame();
+    }
   }
 }
 
+void CastRemotingSender::ConsumeDataChunk(uint32_t offset, uint32_t size,
+                                          uint32_t total_payload_size) {
+  DCHECK(cast_environment_->CurrentlyOn(media::cast::CastEnvironment::MAIN));
+
+  // If |total_payload_size| has changed, resize the data string. If it has not
+  // changed, the following statement will be a no-op.
+  next_frame_data_.resize(total_payload_size);
+
+  do {
+    if (!pipe_.is_valid()) {
+      VLOG(1) << SENDER_SSRC << "Data pipe handle no longer valid.";
+      break;
+    }
+
+    if (offset + size > total_payload_size) {
+      LOG(DFATAL)
+          << "BUG: offset + size > total_payload_size ("
+          << offset << " + " << size << " > " << total_payload_size << ')';
+      break;
+    }
+
+    uint32_t num_bytes = size;
+    const MojoResult result = mojo::ReadDataRaw(
+        pipe_.get(), base::string_as_array(&next_frame_data_) + offset,
+        &num_bytes, MOJO_READ_DATA_FLAG_ALL_OR_NONE);
+    if (result != MOJO_RESULT_OK) {
+      LOG(DFATAL) << "BUG: Read from data pipe failed (" << result << ')';
+      break;
+    }
+
+    return;  // Success.
+  } while (false);
+
+  // If this point is reached, there was a fatal error. Shut things down and run
+  // the error callback.
+  pipe_.reset();
+  binding_.Close();
+  error_callback_.Run();
+}
+
 void CastRemotingSender::SendFrame() {
   DCHECK(cast_environment_->CurrentlyOn(media::cast::CastEnvironment::MAIN));

[xjz, 2016/09/15 18:01:49]

nit: DCHECK_NE(flow_control_state_, CONSUME_PAUSED);

[miu, 2016/09/16 16:39:28]

Done.

 
+  // If there would be too many frames in-flight, suspend consuming more input
+  // from the Mojo data pipes until one or more frames are ACKed.
   if (NumberOfFramesInFlight() >= media::cast::kMaxUnackedFrames) {
-    // TODO(xjz): Delay the sending of this frame and stop reading the next
-    // frame data.
+    binding_.PauseIncomingMethodCallProcessing();
+    flow_control_state_ = CONSUME_PAUSED;
+    VLOG(1) << SENDER_SSRC
+            << "Now CONSUME_PAUSED because too many frames are in flight.";
     return;
   }
 
   VLOG(2) << SENDER_SSRC
           << "About to send another frame: last_sent=" << last_sent_frame_id_
           << ", latest_acked=" << latest_acked_frame_id_;
 
   const media::cast::FrameId frame_id = last_sent_frame_id_ + 1;
   const bool is_first_frame_to_be_sent = last_send_time_.is_null();
 
   base::TimeTicks last_frame_reference_time = last_send_time_;
   last_send_time_ = cast_environment_->Clock()->NowTicks();
   last_sent_frame_id_ = frame_id;
   // If this is the first frame about to be sent, fake the value of
   // |latest_acked_frame_id_| to indicate the receiver starts out all caught up.
   // Also, schedule the periodic frame re-send checks.
   if (is_first_frame_to_be_sent)
     ScheduleNextResendCheck();
 
   DVLOG(3) << "Sending remoting frame, id = " << frame_id;
 
   media::cast::EncodedFrame remoting_frame;
   remoting_frame.frame_id = frame_id;
-  remoting_frame.dependency =
-      (is_first_frame_to_be_sent || last_frame_was_canceled_)
+  if (flow_control_state_ == RESTARTING) {
+    remoting_frame.dependency = media::cast::EncodedFrame::KEY;
+    flow_control_state_ = CONSUMING;
+  } else {
+    remoting_frame.dependency = is_first_frame_to_be_sent
           ? media::cast::EncodedFrame::KEY
           : media::cast::EncodedFrame::DEPENDENT;

[xjz, 2016/09/15 18:01:49]

When |is_first_frame_to_be_send| is true, |flow_control_state_| seems always be
RESTARTING. So maybe:
remoting_frame.dependency = (flow_control_state_ == RESTARTING ?
                                ? media::cast::EncodedFrame::KEY
                                : media::cast::EncodedFrame::DEPENDENT);
flow_control_state_ = CONSUMING;

[miu, 2016/09/16 16:39:28]

Good point. Simplified this a bit, but not exactly as you suggested. PTAL.

+  }
   remoting_frame.referenced_frame_id =
       remoting_frame.dependency == media::cast::EncodedFrame::KEY
           ? frame_id
           : frame_id - 1;
   remoting_frame.reference_time = last_send_time_;
   media::cast::RtpTimeTicks last_frame_rtp_timestamp;
   if (is_first_frame_to_be_sent) {
     last_frame_reference_time = remoting_frame.reference_time;
     last_frame_rtp_timestamp =
         media::cast::RtpTimeTicks() - media::cast::RtpTimeDelta::FromTicks(1);
@@ skipping 18 matching lines @@
   remoting_event->media_type =
       is_audio_ ? media::cast::AUDIO_EVENT : media::cast::VIDEO_EVENT;
   remoting_event->rtp_timestamp = remoting_frame.rtp_timestamp;
   remoting_event->frame_id = frame_id;
   remoting_event->size = remoting_frame.data.length();
   remoting_event->key_frame =
       remoting_frame.dependency == media::cast::EncodedFrame::KEY;
   cast_environment_->logger()->DispatchFrameEvent(std::move(remoting_event));
 
   RecordLatestFrameTimestamps(frame_id, remoting_frame.rtp_timestamp);
-  last_frame_was_canceled_ = false;
 
   transport_->InsertFrame(ssrc_, remoting_frame);
 }
 
-void CastRemotingSender::CancelFramesInFlight() {
+void CastRemotingSender::CancelInFlightData() {
   DCHECK(cast_environment_->CurrentlyOn(media::cast::CastEnvironment::MAIN));
 
   base::STLClearObject(&next_frame_data_);
 
   if (latest_acked_frame_id_ < last_sent_frame_id_) {
     std::vector<media::cast::FrameId> frames_to_cancel;
     do {
       ++latest_acked_frame_id_;
       frames_to_cancel.push_back(latest_acked_frame_id_);
     } while (latest_acked_frame_id_ < last_sent_frame_id_);
     transport_->CancelSendingFrames(ssrc_, frames_to_cancel);
   }
 

[xjz, 2016/09/15 18:01:49]

I might miss one thing here. We need to increase the |last_sent_frame_id_| by
one if there is one frame stored in |next_frame_data_|. And when this is called,
is it possible that there are still some frames stored in Mojo data pipeline? If
yes, we may need to increase |last_sent_frame_id_| by that number too, since we
need to match the frame_id in renderer. As we discussed before, another solution
to prevent this mismatching is to pass the ID from renderer in this function...

[miu, 2016/09/16 16:39:28]

As discussed, we shouldn't be crossing abstraction layers (i.e., protocol
frame_ids should not be visible to the application layer). However, in
discussing this, we realized we can't just blanketly call CancelSendingFrames()
if it's possible any packet made it to the network already.

So, for now, I'm going to comment-out this code with a TODO and file a crbug
(http://crbug.com/647423) for follow-up. Basically, it's okay to not cancel
frames from the transport. However, there is likely to be a large performance
gain from adding an optimization to do so when it doesn't break the protocol
design constraints.

-  last_frame_was_canceled_ = true;
+  flow_control_state_ = RESTARTING;
+  VLOG(1) << SENDER_SSRC
+          << "Now RESTARTING because in-flight data was just canceled.";
 }
 
 }  // namespace cast