[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::WriteBuffers::WriteBuffers() {}
+
+CronetBidirectionalStream::WriteBuffers::~WriteBuffers() {}
+
+void CronetBidirectionalStream::WriteBuffers::Clear() {
+  write_buffer_list.clear();
+  write_buffer_len_list.clear();
+}
+
+void CronetBidirectionalStream::WriteBuffers::AppendBuffer(
+    const scoped_refptr<net::IOBuffer>& buffer,
+    int buffer_size) {
+  write_buffer_list.push_back(buffer);
+  write_buffer_len_list.push_back(buffer_size);
+}
+
+bool CronetBidirectionalStream::WriteBuffers::Empty() const {
+  return write_buffer_list.empty();
+}
+
 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),
+      pending_write_data_(new WriteBuffers()),
+      sending_write_data_(new WriteBuffers()),
       delegate_(delegate) {}
 
 CronetBidirectionalStream::~CronetBidirectionalStream() {
   DCHECK(environment_->IsOnNetworkThread());
 }
 
 int CronetBidirectionalStream::Start(const char* url,
                                      int priority,
                                      const char* method,
                                      const net::HttpRequestHeaders& headers,
@@ skipping 39 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;
 }
 
+void CronetBidirectionalStream::Flush() {
+  environment_->PostToNetworkThread(
+      FROM_HERE, base::Bind(&CronetBidirectionalStream::FlushOnNetworkThread,
+                            base::Unretained(this)));
+}
+
 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(STARTED, write_state_);
+  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(STARTED, read_state_);
   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(READING, read_state_);
   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(WRITING, write_state_);
+  write_state_ = WAITING_FOR_FLUSH;
+  for (scoped_refptr<net::IOBuffer> buffer :

[xunjieli, 2016/06/14 14:29:34]

Does const & work here? I am not sure doing this way will add an extra ref.

[mef, 2016/06/14 19:54:30]

Done.

+       sending_write_data_->write_buffer_list) {
+    delegate_->OnDataSent(buffer->data());
+  }
+  sending_write_data_->Clear();
+  // Send data flushed while other data was sending.
+  if (!pending_write_data_->Empty()) {

[xunjieli, 2016/06/14 14:29:34]

I am not exactly sure if we should auto flush |pending_write_data_|. Andrei,
what do you think?

[kapishnikov, 2016/06/14 14:41:58]

Can it result in flushing data that were written but not actually flushed by the
client? E.g., if the client calls "write", "flush", "write" while the first
write is still in progress, will the second write  be automatically flushed?

[kapishnikov, 2016/06/14 14:50:25]

Sorry Helen, I didn't see your comment, I think we should not autoflush
|pending_write_data_| since it introduces non-deterministic behavior. E.g. let's
say a developer forgot to add flush but ran the tests on a slow networks.
Everything seems to work. Later, when the app is deployed on a faster network,
it can start failing or show flaky behavior. I think we should avoid this type
of scenarios.

[mef, 2016/06/14 19:54:30]

Done.

+    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 18 matching lines @@
   }
   OnDataRead(bytes_read);
 }
 
 void CronetBidirectionalStream::WriteDataOnNetworkThread(
     scoped_refptr<net::WrappedIOBuffer> write_buffer,
     int buffer_size,
     bool end_of_stream) {
   DCHECK(environment_->IsOnNetworkThread());
   DCHECK(write_buffer);
-  DCHECK(!write_buffer_);
-  if (write_state_ != WAITING_FOR_WRITE) {
-    DLOG(ERROR) << "Unexpected Write Data in write_state " << write_state_;
+  DCHECK(!write_end_of_stream_);
+  if (!bidi_stream_ || write_end_of_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;
   }
+  pending_write_data_->AppendBuffer(write_buffer, 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_->Empty()) {
+    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;

[xunjieli, 2016/06/14 14:29:34]

Maybe move this line inside of "if (!request_headers_sent)" block.

[mef, 2016/06/14 19:54:30]

Done.

+  // If previous send is not done, or there is nothing to flush, then exit.
+  if (write_state_ == WRITING)
+    return;
+  DCHECK(!pending_write_data_->Empty());
+  DCHECK(sending_write_data_->Empty());
   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(pending_write_data_);
+  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