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_(STARTING),
       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_)));

[dcheng, 2016/09/20 08:20:12]

Does set_connection_error(sender->error_callback_) work?

[miu, 2016/09/21 03:15:50]

Whoops! Left-overs, ya know? Done.

 }
 
 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));
+
+  // Method calls from Mojo should not be happening when in the paused state.
+  DCHECK_NE(flow_control_state_, CONSUME_PAUSED);

[dcheng, 2016/09/20 08:20:12]

why is that? What stops these messages from being forwarded when we're in
CONSUME_PAUSED?

[miu, 2016/09/21 03:15:50]

Short answer: We call binding_.PauseIncomingMethodCallProcessing(). This
mechanism is comprehensively unit tested
(CastRemotingSenderTest.ManagesFlowControl).

Long answer:

We transition into the CONSUME_PAUSED state when we've run out of buffer
capacity in the network transport. In this state, CastRemotingSender must stop
consuming from the mojo data pipe because there is nowhere for the data to go.

However, the other end (in the render process) will continue to push more data
into the data pipe, and then make calls on the interface to instruct
CastRemotingSender to pull the data out from the pipe. Therefore, when we
transition into the CONSUME_PAUSED state, we also tell Mojo to pause incoming
method calls. This is a super-convenient Mojo feature: CastRemotingSender isn't
being told to consume from the data pipe and so the data will stay there until
we are ready for it.

Once some data has been sent over the network, and we have room again in the
network transport buffers, we tell Mojo to resume incoming method calls. This
will cause CastRemotingSender to resume pulling data out of the data pipe again.

One other concern is whether there is an upper-bound to how many mojo method
calls will be queued-up. There is! It is bound because the data pipe will
eventually fill to capacity. Once the render process cannot push more data into
the data pipe, it will also stop making calls on the interface to have
CastRemotingSender try to consume from the data pipe. It is simply forced to
wait until it sees the data pipe start draining again.

All in all, a pretty natural way to handle flow control. :-)

[dcheng, 2016/09/22 02:29:15]

OK,thanks, I just wanted to make sure we weren't relying on the renderer to stop
making calls and that we had something browser-side.

+
+  // 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)

[dcheng, 2016/09/20 08:20:12]

We probably shouldn't dcheck here, this is coming from the source, which
(presumably) usually lives in a renderer process.

[miu, 2016/09/21 03:15:50]

Relaxed to ERROR level.

+          << "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) {

[dcheng, 2016/09/20 08:20:12]

Is it possible for MOJO_RESULT_OUT_OF_RANGE to be a valid return value here
(e.g. no data is read because the data is not all in the pipe yet) here?

[miu, 2016/09/21 03:15:50]

This should never happen. The render process is always supposed to push data
into the data pipe before calling the interface methods that would have
CastRemotingSender pull from the data pipe. So, the data better be there.

[dcheng, 2016/09/22 02:29:15]

Right, it's not clear to me if there's synchronization between a data pipe and a
message pipe though.

In mojo, unless the interfaces are associated, there's no ordering guarantee.
e.g. if you have

interface_one->Method();
interface_two->Method();

The calls could arrive in either order.

I'm concerned that the same is true for data pipes: if we push data into the
data pipe and then call this interface method, what synchronization guarantees
does mojo provide? If it's consistent with how message pipes for interfaces
work, I'd expect no ordering guarantees.

[miu, 2016/09/22 08:39:14]

Good catch on this, and rockot@ confirmed.

I re-worked the "flow control" logic a bit, and added a queue of retryable input
operations. Now, whenever ConsumeDataChunk() is called, the internal impl
actually checks whether the data pipe has the data. If not, it'll retry at a
later point. I used a mojo::Watcher to signal when data becomes available, to
retry the Consume operation.

Because I had to maintain my own queue, I no longer use Mojo's
Pause/ResumeMethodCallProcessing() feature. However, my earlier comments on why
the input queue will not grow unbounded are still valid.

I also added some extra unit tests to confirm all this new functionality is
working.

+      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));
 
+  // Method calls from Mojo should not be happening when in the paused state.
+  DCHECK_NE(flow_control_state_, CONSUME_PAUSED);
+
+  // 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_)
-          ? media::cast::EncodedFrame::KEY
-          : media::cast::EncodedFrame::DEPENDENT;
+  if (flow_control_state_ == STARTING) {
+    remoting_frame.dependency = media::cast::EncodedFrame::KEY;
+    flow_control_state_ = CONSUMING;
+  } else {
+    DCHECK(!is_first_frame_to_be_sent);
+    remoting_frame.dependency = media::cast::EncodedFrame::DEPENDENT;
+  }
   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));
 
+  // Method calls from Mojo should not be happening when in the paused state.
+  DCHECK_NE(flow_control_state_, CONSUME_PAUSED);
+
   base::STLClearObject(&next_frame_data_);
 
+  // TODO(miu): The following code is something we want to do as an
+  // optimization. However, as-is, it's not quite correct. We can only cancel
+  // frames where no packets have actually hit the network yet. Said another
+  // way, we can only cancel frames the receiver has definitely not seen any
+  // part of (including kickstarting!). http://crbug.com/647423
+#if 0
   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);
   }
+#endif
 
-  last_frame_was_canceled_ = true;
+  flow_control_state_ = STARTING;
+  VLOG(1) << SENDER_SSRC
+          << "Now re-STARTING because in-flight data was just canceled.";
 }
 
 }  // namespace cast