Add QuicDataChannel and QuicDataTransport classes

webrtc/api/quicdatachannel.cc

+/*
+ *  Copyright 2016 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#include "webrtc/api/quicdatachannel.h"
+
+#include "webrtc/base/bind.h"
+#include "webrtc/base/bytebuffer.h"
+#include "webrtc/base/copyonwritebuffer.h"
+#include "webrtc/base/logging.h"
+#include "webrtc/p2p/quic/quictransportchannel.h"
+#include "webrtc/p2p/quic/reliablequicstream.h"
+
+namespace webrtc {
+
+void WriteQuicDataChannelMessageHeader(int data_channel_id,
+                                       uint64_t message_id,
+                                       rtc::CopyOnWriteBuffer* header) {
+  RTC_DCHECK(header);
+  // 64-bit varints require at most 10 bytes (7*10 == 70), and 32-bit varints
+  // require at most 5 bytes (7*5 == 35).
+  size_t max_length = 15;
+  rtc::ByteBufferWriter byte_buffer(nullptr, max_length,
+                                    rtc::ByteBuffer::ByteOrder::ORDER_HOST);
+  byte_buffer.WriteUVarint(data_channel_id);
+  byte_buffer.WriteUVarint(message_id);
+  header->SetData(byte_buffer.Data(), byte_buffer.Length());
+}
+
+bool ParseQuicDataMessageHeader(const char* data,
+                                size_t len,
+                                int* data_channel_id,
+                                uint64_t* message_id,
+                                size_t* bytes_read) {
+  RTC_DCHECK(data_channel_id);
+  RTC_DCHECK(message_id);
+  RTC_DCHECK(bytes_read);
+
+  rtc::ByteBufferReader byte_buffer(data, len, rtc::ByteBuffer::ORDER_HOST);
+  uint64_t dcid;
+  if (!byte_buffer.ReadUVarint(&dcid)) {
+    LOG(LS_ERROR) << "Could not read the data channel ID";
+    return false;
+  }
+  *data_channel_id = dcid;
+  if (!byte_buffer.ReadUVarint(message_id)) {
+    LOG(LS_ERROR) << "Could not read message ID for data channel "
+                  << *data_channel_id;
+    return false;
+  }
+  size_t remaining_bytes = byte_buffer.Length();
+  *bytes_read = len - remaining_bytes;
+  return true;
+}
+
+QuicDataChannel::QuicDataChannel(rtc::Thread* signaling_thread,
+                                 rtc::Thread* worker_thread,
+                                 const std::string& label,
+                                 const DataChannelInit& config)
+    : signaling_thread_(signaling_thread),
+      worker_thread_(worker_thread),
+      id_(config.id),
+      state_(kConnecting),
+      buffered_amount_(0),
+      next_message_id_(0),
+      label_(label),
+      protocol_(config.protocol) {}
+
+QuicDataChannel::~QuicDataChannel() {}
+
+void QuicDataChannel::RegisterObserver(DataChannelObserver* observer) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  observer_ = observer;
+}
+
+void QuicDataChannel::UnregisterObserver() {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  observer_ = nullptr;
+}
+
+bool QuicDataChannel::Send(const DataBuffer& buffer) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (state_ != kOpen) {
+    LOG(LS_ERROR) << "QUIC data channel " << id_
+                  << " is not open so cannot send.";
+    return false;
+  }
+  return worker_thread_->Invoke<bool>(
+      rtc::Bind(&QuicDataChannel::Send_w, this, buffer));
+}
+
+bool QuicDataChannel::Send_w(const DataBuffer& buffer) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+
+  // Encode and send the header containing the data channel ID and message ID.
+  rtc::CopyOnWriteBuffer header;
+  WriteQuicDataChannelMessageHeader(id_, ++next_message_id_, &header);
+  RTC_DCHECK(quic_transport_channel_);
+  cricket::ReliableQuicStream* stream =
+      quic_transport_channel_->CreateQuicStream();
+  RTC_DCHECK(stream);
+
+  // Send the header with a FIN if the message is empty.
+  bool header_fin = (buffer.size() == 0);
+  rtc::StreamResult header_result =
+      stream->Write(header.data<char>(), header.size(), header_fin);
+
+  if (header_result == rtc::SR_BLOCK) {
+    // The header is write blocked but we should try sending the message. Since
+    // the ReliableQuicStream queues data in order, if the header is write
+    // blocked then the message will be write blocked. Otherwise if the message
+    // is sent then the header is sent.
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " header is write blocked for QUIC data channel " << id_;
+  } else if (header_result != rtc::SR_SUCCESS) {
+    LOG(LS_ERROR) << "Stream " << stream->id()
+                  << " failed to write header for QUIC data channel " << id_
+                  << ". Unexpected error " << header_result;
+    return false;
+  }
+
+  // If the message is not empty, then send the message with a FIN.
+  bool message_fin = true;
+  rtc::StreamResult message_result =
+      header_fin ? header_result : stream->Write(buffer.data.data<char>(),
+                                                 buffer.size(), message_fin);
+
+  if (message_result == rtc::SR_SUCCESS) {
+    // The message is sent and we don't need this QUIC stream.
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " successfully wrote message for QUIC data channel " << id_;
+    stream->Close();
+    return true;
+  }
+  // TODO(mikescarlett): Register the ReliableQuicStream's priority to the
+  // QuicWriteBlockedList so that the QUIC session doesn't drop messages when
+  // the QUIC transport channel becomes unwritable.
+  if (message_result == rtc::SR_BLOCK) {
+    // The QUIC stream is write blocked, so the message is queued by the QUIC
+    // session. If this is due to the QUIC not being writable, it will be sent
+    // once QUIC becomes writable again. Otherwise it may be due to exceeding
+    // the QUIC flow control limit, in which case the remote peer's QUIC session
+    // will tell the QUIC stream to send more data.
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " message is write blocked for QUIC data channel " << id_;
+    SetBufferedAmount_w(buffered_amount_ + stream->queued_data_bytes());
+    stream->SignalQueuedBytesWritten.connect(
+        this, &QuicDataChannel::OnQueuedBytesWritten);
+    write_blocked_quic_streams_[stream->id()] = stream;
+    // The QUIC stream will be removed from |write_blocked_quic_streams_| once
+    // it closes.
+    stream->SignalClosed.connect(this,
+                                 &QuicDataChannel::OnWriteBlockedStreamClosed);
+    return true;
+  }
+  LOG(LS_ERROR) << "Stream " << stream->id()
+                << " failed to write message for QUIC data channel " << id_
+                << ". Unexpected error: " << message_result;
+  return false;
+}
+
+void QuicDataChannel::OnQueuedBytesWritten(net::QuicStreamId stream_id,
+                                           uint64_t queued_bytes_written) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  SetBufferedAmount_w(buffered_amount_ - queued_bytes_written);
+  const auto& kv = write_blocked_quic_streams_.find(stream_id);
+  if (kv == write_blocked_quic_streams_.end()) {
+    RTC_DCHECK(false);
+    return;
+  }
+  cricket::ReliableQuicStream* stream = kv->second;
+  // True if the QUIC stream is done sending data.
+  if (stream->fin_sent()) {
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " successfully wrote data for QUIC data channel " << id_;
+    stream->Close();
+  }
+}
+
+void QuicDataChannel::SetBufferedAmount_w(uint64_t buffered_amount) {
+  buffered_amount_ = buffered_amount;
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_, rtc::Bind(&QuicDataChannel::OnBufferedAmountChange_s,
+                                   this, buffered_amount));
+}
+
+void QuicDataChannel::Close() {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (state_ == kClosed || state_ == kClosing) {
+    return;
+  }
+  LOG(LS_INFO) << "Closing QUIC data channel.";
+  SetState_s(kClosing);
+  worker_thread_->Invoke<void>(rtc::Bind(&QuicDataChannel::Close_w, this));
+  SetState_s(kClosed);
+}
+
+void QuicDataChannel::Close_w() {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  for (auto& kv : incoming_quic_messages_) {
+    Message& message = kv.second;
+    cricket::ReliableQuicStream* stream = message.stream;
+    stream->Close();
+  }
+
+  for (auto& kv : write_blocked_quic_streams_) {
+    cricket::ReliableQuicStream* stream = kv.second;
+    stream->Close();
+  }
+}
+
+bool QuicDataChannel::SetTransportChannel(
+    cricket::QuicTransportChannel* channel) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+
+  if (!channel) {
+    LOG(LS_ERROR) << "|channel| is NULL. Cannot set transport channel.";
+    return false;
+  }
+  if (quic_transport_channel_) {
+    if (channel == quic_transport_channel_) {
+      LOG(LS_WARNING) << "Ignoring duplicate transport channel.";
+      return true;
+    }
+    LOG(LS_ERROR) << "|channel| does not match existing transport channel.";
+    return false;
+  }
+
+  quic_transport_channel_ = channel;
+  LOG(LS_INFO) << "Setting QuicTransportChannel for QUIC data channel " << id_;
+  worker_thread_->Invoke<void>(
+      rtc::Bind(&QuicDataChannel::SetTransportChannel_w, this));
+  return true;
+}
+
+void QuicDataChannel::SetTransportChannel_w() {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  quic_transport_channel_->SignalReadyToSend.connect(
+      this, &QuicDataChannel::OnReadyToSend);
+  quic_transport_channel_->SignalClosed.connect(
+      this, &QuicDataChannel::OnConnectionClosed);
+  if (quic_transport_channel_->writable()) {
+    invoker_.AsyncInvoke<void>(
+        signaling_thread_,
+        rtc::Bind(&QuicDataChannel::SetState_s, this, kOpen));
+  }

[pthatcher1, 2016/04/29 22:04:17]

Hmmm.... I think what we want to do here is have SetTransportChannel_w return
the state it should be set to, and then have SetTransportChannel set it to the
state.  That way, it happens immediately and writable() will be read on the
worker thread and state_ accessed on the signaling thread.

[mikescarlett, 2016/04/29 22:11:04]

Good idea. Done.

+}
+
+void QuicDataChannel::OnIncomingMessage(Message&& message) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(message.stream);
+  if (!observer_) {
+    LOG(LS_WARNING) << "QUIC data channel " << id_
+                    << " received a message but has no observer.";
+    message.stream->Close();
+    return;
+  }
+  // A FIN is received if the message fits into a single QUIC stream frame and
+  // the remote peer is done sending.
+  if (message.stream->fin_received()) {
+    LOG(LS_INFO) << "Stream " << message.stream->id()
+                 << " has finished receiving data for QUIC data channel "
+                 << id_;
+    DataBuffer final_message(message.buffer, false);
+    invoker_.AsyncInvoke<void>(signaling_thread_,
+                               rtc::Bind(&QuicDataChannel::OnMessage_s, this,
+                                         std::move(final_message)));
+    message.stream->Close();
+    return;
+  }
+  // Otherwise the message is divided across multiple QUIC stream frames, so
+  // queue the data. OnDataReceived() will be called each time the remaining
+  // QUIC stream frames arrive.
+  LOG(LS_INFO) << "QUIC data channel " << id_
+               << " is queuing incoming data for stream "
+               << message.stream->id();
+  incoming_quic_messages_[message.stream->id()] = std::move(message);
+  message.stream->SignalDataReceived.connect(this,
+                                             &QuicDataChannel::OnDataReceived);
+  // The QUIC stream will be removed from |incoming_quic_messages_| once it
+  // closes.
+  message.stream->SignalClosed.connect(
+      this, &QuicDataChannel::OnIncomingQueuedStreamClosed);
+}
+
+void QuicDataChannel::OnDataReceived(net::QuicStreamId stream_id,
+                                     const char* data,
+                                     size_t len) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(data);
+  const auto& kv = incoming_quic_messages_.find(stream_id);
+  if (kv == incoming_quic_messages_.end()) {
+    RTC_DCHECK(false);
+    return;
+  }
+  Message& message = kv->second;
+  cricket::ReliableQuicStream* stream = message.stream;
+  rtc::CopyOnWriteBuffer& received_data = message.buffer;
+  // If the QUIC stream has not received a FIN, then the remote peer is not
+  // finished sending data.
+  if (!stream->fin_received()) {
+    received_data.AppendData(data, len);
+    return;
+  }
+  // Otherwise we are done receiving and can provide the data channel observer
+  // with the message.
+  LOG(LS_INFO) << "Stream " << stream_id
+               << " has finished receiving data for QUIC data channel " << id_;
+  received_data.AppendData(data, len);
+  DataBuffer final_message(std::move(received_data), false);
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_,
+      rtc::Bind(&QuicDataChannel::OnMessage_s, this, std::move(final_message)));
+  // Once the stream is closed, OnDataReceived will not fire for the stream.
+  stream->Close();
+}
+
+void QuicDataChannel::OnReadyToSend(cricket::TransportChannel* channel) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(channel == quic_transport_channel_);
+  LOG(LS_INFO) << "QuicTransportChannel is ready to send";
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_, rtc::Bind(&QuicDataChannel::SetState_s, this, kOpen));
+}
+
+void QuicDataChannel::OnWriteBlockedStreamClosed(net::QuicStreamId stream_id,
+                                                 int error) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  LOG(LS_VERBOSE) << "Write blocked stream " << stream_id << " is closed.";
+  write_blocked_quic_streams_.erase(stream_id);
+}
+
+void QuicDataChannel::OnIncomingQueuedStreamClosed(net::QuicStreamId stream_id,
+                                                   int error) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  LOG(LS_VERBOSE) << "Incoming queued stream " << stream_id << " is closed.";
+  incoming_quic_messages_.erase(stream_id);
+}
+
+void QuicDataChannel::OnConnectionClosed() {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  invoker_.AsyncInvoke<void>(signaling_thread_,
+                             rtc::Bind(&QuicDataChannel::Close, this));
+}
+
+void QuicDataChannel::OnMessage_s(const DataBuffer& received_data) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (observer_) {
+    observer_->OnMessage(received_data);
+  }
+}
+
+void QuicDataChannel::SetState_s(DataState state) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (state_ == state || state_ == kClosed) {
+    return;
+  }
+  if (state_ == kClosing && state != kClosed) {
+    return;
+  }
+  LOG(LS_INFO) << "Setting state to " << state << " for QUIC data channel "
+               << id_;
+  state_ = state;
+  if (observer_) {
+    observer_->OnStateChange();
+  }
+}
+
+void QuicDataChannel::OnBufferedAmountChange_s(uint64_t buffered_amount) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (observer_) {
+    observer_->OnBufferedAmountChange(buffered_amount);
+  }
+}
+
+size_t QuicDataChannel::GetNumWriteBlockedStreams() const {
+  return write_blocked_quic_streams_.size();
+}
+
+size_t QuicDataChannel::GetNumIncomingStreams() const {
+  return incoming_quic_messages_.size();
+}
+
+}  // namespace webrtc