[Cronet] Coalesce small buffers into single QUIC packet in GRPC on iOS.

components/cronet/ios/cronet_bidirectional_stream.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 "components/cronet/ios/cronet_bidirectional_stream.h"
 
 #include <memory>
 #include <string>
 #include <vector>
 
@@ skipping 14 matching lines @@
 #include "net/http/http_transaction_factory.h"
 #include "net/http/http_util.h"
 #include "net/spdy/spdy_header_block.h"
 #include "net/ssl/ssl_info.h"
 #include "net/url_request/http_user_agent_settings.h"
 #include "net/url_request/url_request_context.h"
 #include "url/gurl.h"
 
 namespace cronet {
 
+CronetBidirectionalStream::PendingWriteData::PendingWriteData() {}
+
+CronetBidirectionalStream::PendingWriteData::~PendingWriteData() {}
+
 CronetBidirectionalStream::CronetBidirectionalStream(
     CronetEnvironment* environment,
     Delegate* delegate)
     : read_state_(NOT_STARTED),
       write_state_(NOT_STARTED),
       write_end_of_stream_(false),
+      request_headers_sent_(false),
+      disable_auto_flush_(false),
+      delay_headers_until_flush_(false),
       environment_(environment),
       delegate_(delegate) {}
 
 CronetBidirectionalStream::~CronetBidirectionalStream() {
   DCHECK(environment_->IsOnNetworkThread());
 }
 
 int CronetBidirectionalStream::Start(const char* url,
                                      int priority,
                                      const char* method,
@@ skipping 40 matching lines @@
   scoped_refptr<net::WrappedIOBuffer> write_buffer(
       new net::WrappedIOBuffer(buffer));
 
   environment_->PostToNetworkThread(
       FROM_HERE,
       base::Bind(&CronetBidirectionalStream::WriteDataOnNetworkThread,
                  base::Unretained(this), write_buffer, count, end_of_stream));
   return true;
 }
 
+bool CronetBidirectionalStream::Flush() {
+  environment_->PostToNetworkThread(
+      FROM_HERE, base::Bind(&CronetBidirectionalStream::FlushOnNetworkThread,
+                            base::Unretained(this)));
+  return true;
+}
+
 void CronetBidirectionalStream::Cancel() {
   environment_->PostToNetworkThread(
       FROM_HERE, base::Bind(&CronetBidirectionalStream::CancelOnNetworkThread,
                             base::Unretained(this)));
 }
 
 void CronetBidirectionalStream::Destroy() {
   // Destroy could be called from any thread, including network thread (if
   // posting task to executor throws an exception), but is posted, so |this|
   // is valid until calling task is complete.
   environment_->PostToNetworkThread(
       FROM_HERE, base::Bind(&CronetBidirectionalStream::DestroyOnNetworkThread,
                             base::Unretained(this)));
 }
 
 void CronetBidirectionalStream::OnStreamReady(bool request_headers_sent) {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(write_state_ == STARTED);
-  write_state_ = WAITING_FOR_WRITE;
-  if (write_end_of_stream_)
+  DCHECK_EQ(write_state_, STARTED);
+  request_headers_sent_ = request_headers_sent;
+  write_state_ = WAITING_FOR_FLUSH;
+  if (write_end_of_stream_) {
+    if (!request_headers_sent) {
+      // If there is no data to write, then just send headers explicitly.
+      bidi_stream_->SendRequestHeaders();
+      request_headers_sent_ = true;
+    }
     write_state_ = WRITING_DONE;
+  }
   delegate_->OnStreamReady();
 }
 
 void CronetBidirectionalStream::OnHeadersReceived(
     const net::SpdyHeaderBlock& response_headers) {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(read_state_ == STARTED);
+  DCHECK_EQ(read_state_, STARTED);

[xunjieli, 2016/06/10 20:31:37]

nit: the expected state should be the first arg.
DCHECK_EQ(STARTED, read_state_);

[mef, 2016/06/10 21:15:14]

Hrm, I think that's true for ASSERT_EQ, but I see DCHECK_EQ both ways. I don't
mind changing though.

[xunjieli, 2016/06/10 22:22:09]

I have been doing that way. But if there are places in the code base that use it
in the other way, then I am not sure. Looking at the header file, DCHECK_EQ does
put the variable in the second argument:
https://cs.chromium.org/chromium/src/base/logging.h?rcl=0&l=746.

[kapishnikov, 2016/06/13 21:21:57]

Found this one: https://bugs.chromium.org/p/chromium/issues/detail?id=58409#c13.
The order of DCHECK_EQ arguments is "expected", "actual".

[mef, 2016/06/14 00:20:38]

Done.

   read_state_ = WAITING_FOR_READ;
   // Get http status code from response headers.
   int http_status_code = 0;
   const auto http_status_header = response_headers.find(":status");
   if (http_status_header != response_headers.end())
     base::StringToInt(http_status_header->second, &http_status_code);
   const char* protocol = "unknown";
   switch (bidi_stream_->GetProtocol()) {
     case net::kProtoHTTP2:
       protocol = "h2";
       break;
     case net::kProtoQUIC1SPDY3:
       protocol = "quic/1+spdy/3";
       break;
     default:
       break;
   }
   delegate_->OnHeadersReceived(response_headers, protocol);
 }
 
 void CronetBidirectionalStream::OnDataRead(int bytes_read) {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(read_state_ == READING);
+  DCHECK_EQ(read_state_, READING);
   read_state_ = WAITING_FOR_READ;
   delegate_->OnDataRead(read_buffer_->data(), bytes_read);
 
   // Free the read buffer.
   read_buffer_ = nullptr;
   if (bytes_read == 0)
     read_state_ = READING_DONE;
   MaybeOnSucceded();
 }
 
 void CronetBidirectionalStream::OnDataSent() {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(write_state_ == WRITING);
-  write_state_ = WAITING_FOR_WRITE;
-  delegate_->OnDataSent(write_buffer_->data());
-  // Free the write buffer.
-  write_buffer_ = nullptr;
+  DCHECK_EQ(write_state_, WRITING);
+  write_state_ = WAITING_FOR_FLUSH;
+  for (scoped_refptr<net::IOBuffer> buffer :
+       sending_write_data_->write_buffer_list) {
+    delegate_->OnDataSent(buffer->data());
+  }
+  sending_write_data_.reset();
+  // Send data flushed while other data was sending.
+  if (flushing_write_data_) {
+    FlushOnNetworkThread();
+    return;
+  }
   if (write_end_of_stream_)
     write_state_ = WRITING_DONE;
   MaybeOnSucceded();
 }
 
 void CronetBidirectionalStream::OnTrailersReceived(
     const net::SpdyHeaderBlock& response_trailers) {
   DCHECK(environment_->IsOnNetworkThread());
   delegate_->OnTrailersReceived(response_trailers);
 }
 
 void CronetBidirectionalStream::OnFailed(int error) {
   DCHECK(environment_->IsOnNetworkThread());
   bidi_stream_.reset();
   read_state_ = write_state_ = ERROR;
   delegate_->OnFailed(error);
 }
 
 void CronetBidirectionalStream::StartOnNetworkThread(
     std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info) {
   DCHECK(environment_->IsOnNetworkThread());
   DCHECK(!bidi_stream_);
   DCHECK(environment_->GetURLRequestContext());
   request_info->extra_headers.SetHeaderIfMissing(
       net::HttpRequestHeaders::kUserAgent, environment_->user_agent());
   bidi_stream_.reset(new net::BidirectionalStream(
       std::move(request_info), environment_->GetURLRequestContext()
                                    ->http_transaction_factory()
                                    ->GetSession(),
-      /*send_request_headers_automatically=*/true, this));
+      !delay_headers_until_flush_, this));
   DCHECK(read_state_ == NOT_STARTED && write_state_ == NOT_STARTED);
   read_state_ = write_state_ = STARTED;
 }
 
 void CronetBidirectionalStream::ReadDataOnNetworkThread(
     scoped_refptr<net::WrappedIOBuffer> read_buffer,
     int buffer_size) {
   DCHECK(environment_->IsOnNetworkThread());
   DCHECK(read_buffer);
   DCHECK(!read_buffer_);
@@ skipping 16 matching lines @@
     OnFailed(bytes_read);
     return;
   }
   OnDataRead(bytes_read);
 }
 
 void CronetBidirectionalStream::WriteDataOnNetworkThread(
     scoped_refptr<net::WrappedIOBuffer> write_buffer,
     int buffer_size,
     bool end_of_stream) {
   DCHECK(environment_->IsOnNetworkThread());

[kapishnikov, 2016/06/13 21:21:57]

Can we add DCHECK here to check that write_end_of_stream_ is 'false'? Maybe we
should call onFailed()?

[mef, 2016/06/14 00:20:37]

Done.

   DCHECK(write_buffer);
-  DCHECK(!write_buffer_);
-  if (write_state_ != WAITING_FOR_WRITE) {
-    DLOG(ERROR) << "Unexpected Write Data in write_state " << write_state_;
+  if (!bidi_stream_) {
+    DLOG(ERROR) << "Unexpected Flush Data in write_state " << write_state_;
     // Invoke OnFailed unless it is already invoked.
     if (write_state_ != ERROR)
       OnFailed(net::ERR_UNEXPECTED);
     return;
   }
+  if (!pending_write_data_)
+    pending_write_data_.reset(new PendingWriteData());
+  pending_write_data_->write_buffer_list.push_back(write_buffer);

[kapishnikov, 2016/06/13 21:21:57]

Can we create a method inside PendingWriteData struct and combine
write_buffer_list.push_back() and write_buffer_len_list.push_back() in this
method? E.g.,
pending_write_data_->AppendBuffer(write_buffer, buffer_size).

I think it will make it more object-oriented.

[mef, 2016/06/14 00:20:38]

Done.

+  pending_write_data_->write_buffer_len_list.push_back(buffer_size);
+  write_end_of_stream_ = end_of_stream;
+  if (!disable_auto_flush_)
+    FlushOnNetworkThread();
+}
+
+void CronetBidirectionalStream::FlushOnNetworkThread() {
+  DCHECK(environment_->IsOnNetworkThread());
+  if (!bidi_stream_)
+    return;
+  // If there is no data to flush, may need to send headers.
+  if (!pending_write_data_) {
+    if (!request_headers_sent_) {
+      request_headers_sent_ = true;
+      bidi_stream_->SendRequestHeaders();
+    }
+    return;
+  }
+  // If request headers are not sent yet, they will be sent with the data.
+  request_headers_sent_ = true;
+  // Move pending data to the flushing list.
+  if (flushing_write_data_) {

[xunjieli, 2016/06/10 20:31:37]

I think it is slightly confusing that |flushing_write_data_| and
|pending_write_data_| can be in three states: (1) null, (2) not-null but empty,
(3) not-null and not empty.

It might be easier if we have two std::vector<WrappedBufferWithSize>, which can
only be empty or not empty, but never null. Can create a WrappedBufferWithSize
which is a subclass of net::WrappedBuffer, but with a size.

WDYT?

[mef, 2016/06/10 21:15:14]

That's an interesting obeservation. With FlushOnNetworkThread doing moving data
intp flushing vector anyway, I wonder whether we actually need 2 separate lists.
I'll think about it.

[mef, 2016/06/14 00:20:38]

Done. I think. :)

+    std::move(pending_write_data_->write_buffer_list.begin(),

[kapishnikov, 2016/06/13 21:21:57]

Maybe a new method in PendingWriteData struct? E.g.,
flushing_write_data_->MoveDataBuffers(pending_write_data_).

[mef, 2016/06/14 00:20:37]

I think we don't need to separate pending_write_data from flusing_write_data as
we always flush everything flushed before explicit Flush or onDataSent.

+              pending_write_data_->write_buffer_list.end(),
+              std::back_inserter(flushing_write_data_->write_buffer_list));
+    std::move(pending_write_data_->write_buffer_len_list.begin(),
+              pending_write_data_->write_buffer_len_list.end(),
+              std::back_inserter(flushing_write_data_->write_buffer_len_list));
+  } else {
+    std::swap(flushing_write_data_, pending_write_data_);
+  }
+  // If previous send is not done, or there is nothing to flush, then exit.
+  if ((write_state_ == WRITING) || !flushing_write_data_)

[kapishnikov, 2016/06/13 21:21:56]

I think the content of flushing_write_data_ can never be NULL here. Should we
add DCHECK instead?

[mef, 2016/06/14 00:20:38]

Done.

+    return;
   write_state_ = WRITING;
-  write_end_of_stream_ = end_of_stream;
-
-  write_buffer_ = write_buffer;
-  bidi_stream_->SendData(write_buffer_.get(), buffer_size, end_of_stream);
+  sending_write_data_.swap(flushing_write_data_);

[kapishnikov, 2016/06/13 21:21:57]

Should we add DCHECK that the content of sending_write_data_ is NULL? I assume
that should always be true here.

[mef, 2016/06/14 00:20:38]

Done.

+  bidi_stream_->SendvData(sending_write_data_->write_buffer_list,
+                          sending_write_data_->write_buffer_len_list,
+                          write_end_of_stream_);
 }
 
 void CronetBidirectionalStream::CancelOnNetworkThread() {
   DCHECK(environment_->IsOnNetworkThread());
   if (!bidi_stream_)
     return;
   read_state_ = write_state_ = CANCELED;
   bidi_stream_.reset();
   delegate_->OnCanceled();
 }
 
 void CronetBidirectionalStream::DestroyOnNetworkThread() {
   DCHECK(environment_->IsOnNetworkThread());
   delete this;
 }
 
 void CronetBidirectionalStream::MaybeOnSucceded() {
   DCHECK(environment_->IsOnNetworkThread());
   if (read_state_ == READING_DONE && write_state_ == WRITING_DONE) {
     read_state_ = write_state_ = SUCCESS;
     bidi_stream_.reset();
     delegate_->OnSucceeded();
   }
 }
 
 }  // namespace cronet