[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_)
+  request_headers_sent_ = request_headers_sent;
+  write_state_ = WAITING_FOR_FLUSH;
+  if (write_end_of_stream_) {
+    if (!request_headers_sent) {

[xunjieli, 2016/06/10 16:30:49]

Should we not check |write_end_stream_|? Currently if it is a GET request and
|delay_headers_until_flush_| is set, we delay the headers until flush() in Java.
I wonder whether we should do the same.

[mef, 2016/06/10 18:36:36]

But that would never come if auto flush is enabled, right?

[xunjieli, 2016/06/10 19:00:01]

You are right. Hopefully we will get rid of auto flush soon, so we don't run
into this case.

+      // 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);
   read_state_ = WAITING_FOR_READ;
   // Get http status code from response headers.
   int http_status_code = 0;
@@ skipping 22 matching lines @@
 
   // 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);

[xunjieli, 2016/06/10 16:30:49]

DCHECK_EQ?

[mef, 2016/06/10 18:36:36]

Done.

-  write_state_ = WAITING_FOR_WRITE;
-  delegate_->OnDataSent(write_buffer_->data());
-  // Free the write buffer.
-  write_buffer_ = nullptr;
+  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_) {
+    SendFlushingWriteData();

[xunjieli, 2016/06/10 16:30:50]

Maybe invoke FlushOnNetworkThread() directly? This is match the Java
implementation more closely. Having one fewer method is easier to manage.

[mef, 2016/06/10 18:36:36]

I think they are different.
FlushOnNetworkThread is posted from app call to Flush() at any time.
SendFlushingWriteData() is called when we are ready to send data that was
accumulated for flushing.

+    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_;
+  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);
+  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_) {

[xunjieli, 2016/06/10 16:30:50]

Also need to check |flushing_write_data_| ?

[mef, 2016/06/10 18:36:36]

Hrm, I think if |flushing_write_data_| is not null, then it means that
FlushOnNetworkThread was called before, and headers will be sent with the 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_) {
+    std::move(pending_write_data_->write_buffer_list.begin(),
+              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 (!sending_write_data_)

[xunjieli, 2016/06/10 19:00:01]

Can this check be just "if (write_state_ != WRITING)"?

[mef, 2016/06/10 19:11:28]

Done.

+    SendFlushingWriteData();
+}
+
+void CronetBidirectionalStream::SendFlushingWriteData() {
+  DCHECK(environment_->IsOnNetworkThread());
+  DCHECK(!sending_write_data_);
+  if (write_state_ != WAITING_FOR_FLUSH) {

[xunjieli, 2016/06/10 16:30:50]

Flush() can be called anytime after OnStreamReady() (even after stream is
destroyed), so we need to handle that.

In FlushOnNetworkThread, maybe check that write_state_ == WAITING_FOR_FLUSH, if
not, just accumulate the pending write data in the flushing write data?

Suggest combining this method with FlushOnNetworkThread() method. It might be
slightly easier to reason. 

[mef, 2016/06/10 18:36:36]

SendFlushingWriteData is similar to nativeWritevData, it never gets called
directly by the app, only from FlushOnNetworkThread or onDataSent when previous
set of buffers has flushed.

[xunjieli, 2016/06/10 19:00:01]

I learned from Andrei that we should re-use routine where possible during the
review of the Android version of this change. It looks like we can combine
SendFlushingWriteData() with the logic in FlushOnNetworkThread(). In
OnDataSent(), we can invoke FlushOnNetworkThread() directly.
FlushOnNetworkThread() will send pending data and/or pending flush data
accordingly. I think it might be a cleaner interface that way.

[xunjieli, 2016/06/10 19:04:38]

In other words, iI am wondering if we can get rid of SendFlushingWriteData(),
and move the logic to FlushOnNetworkThread(). The SendFlushingWriteData() method
name is a bit unclear, and it seems that it is doing what FlushOnNetworkThread()
should do. If it is, making it into a separate method might be confusing.

[mef, 2016/06/10 19:11:28]

I don't think we can though. FlushOnNetworkThread moves all pending write data
into flushing_write_data_ and SendFlushingWriteData sends flushing_write_data_
if any.

We don't want to flush pending data from onDataSent as it maybe incomplete (e.g.
app written 2 buffers out of 5 and hasn't called Flush yet).

[mef, 2016/06/10 20:03:48]

Per offline discussion this is current behavior on Android, and should be ok, so
I've changed iOS implementation to match.

+    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 (!flushing_write_data_)
+    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_);
+  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